diff --git a/TODO.md b/TODO.md index 07bd2f2c1..8ce65adea 100644 --- a/TODO.md +++ b/TODO.md @@ -22,7 +22,7 @@ - EC2 - Support extended features for `run-instance` - [DONE] Volume resize of AMI. Note `nbd` does not support live resize events, instance needs to be stopped, resized, and started. - Confirm cloud-init will resize volume on boot and configured correctly. - - Attach additional volumes + - [DONE] Attach additional volumes - Attach to VPC / Security group (required Open vSwitch implementation) - only allow hive to use x% of cpu and memory. dont allow allocating 100% of resources. - placement group instances, allow spreading instance group between nodes. @@ -34,43 +34,21 @@ - [DONE] change all log into slog - `nbd` does not support resizing disks live. Requires instance to be stopped, boot/root volume, or attached volume will need to be resized, and instance started again. Limitation for Hive v1 using NBD, aiming to resolve with `vhost-user-blk` to create a `virtio-blk` device for extended functionality and performance. -## EC2 attach-volume +## [DONE] EC2 attach-volume ```sh aws ec2 attach-volume --volume-id vol-1234567890abcdef0 --instance-id i-01474ef662b89480 --device /dev/sdf ``` -- Validate volume exists, e.g create-volume -- Push attach-volume request via NATS to the node running the specified instance - Run nbdkit for new volume, retrieve host:port or UNIX socket pathname, append volume to instance state, so will mount/attach on instance stop/start. -- Issue a command via QMP for the instance, to attach the volume, e.g -- Confirm if --device compatible, since NBD will export the drive, likely the host OS will decide the /dev/volume attributes. -- Volume available to mount within the guest OS. Volume meta-data attached to instance.json/NATS KV and available if instance stop/started. +- [DONE] Validate volume exists, e.g create-volume — `GetVolumeConfig()` checks volume exists and state == "available" +- [DONE] Push attach-volume request via NATS to the node running the specified instance — gateway sends to `ec2.cmd.{instanceId}`, daemon subscribes per-instance +- [DONE] Run viperblock NBD server for new volume, retrieve host:port or UNIX socket pathname, append volume to instance state — `ebs.mount` via NATS returns NBD URI, volume appended to `EBSRequests`, persisted to JetStream so mounts survive stop/start +- [DONE] Issue a command via QMP for the instance, to attach the volume — `blockdev-add` (NBD backend) + `device_add` (virtio-blk-pci) with three-phase rollback +- [DONE] --device auto-assignment — if `--device` omitted, `nextAvailableDevice()` scans EBSRequests + BlockDeviceMappings to assign next `/dev/sd[f-p]`. Guest sees `/dev/vd*` with virtio drivers. +- [DONE] Volume available to mount within the guest OS. Volume meta-data attached to instance.json/NATS KV and available if instance stop/started. -```json -{ - "execute": "blockdev-add", - "arguments": { - "node-name": "nbd0", - "driver": "nbd", - "server": { "type": "unix", "path": "/run/nbd/vol0.sock" }, - "export": "", // Note, export volume should be blank for viperblock - "read-only": false - } -} -``` - -Once attached, QMP event to add the device. Likely --device argument can be specified here for /dev/block device. If --device missing, need to loop through the instance attached volumes and determine an id, e.g /dev/sda used already, /dev/sdb available. - -```json -{ - "execute": "device_add", - "arguments": { - "driver": "virtio-blk-pci", - "id": "vdisk1", - "drive": "nbd0" - } -} -``` +See `docs/development/features/attach-volume-testing.md` for manual testing guide. +See `docs/development/feature/attach-volume.md` for implementation details. ## EC2 detach-volume @@ -191,7 +169,7 @@ Instance `dmesg`: | `modify-volume` | `--volume-id`, `--size`, `--volume-type`, `--iops` | `--throughput`, `--dry-run`, `--multi-attach-enabled` | Volume must exist | NATS `ec2.ModifyVolume` → daemon sends resize request to viperblock → NBD does not support live resize, instance must be stopped → returns modification state | 1. Increase volume size
2. Modify volume type
3. Decrease size (error - not supported)
4. Modify attached volume (requires stop/start) | **DONE** | | `create-volume` | `--size`, `--availability-zone`, `--volume-type` (gp3 only) | `--iops`, `--encrypted`, `--snapshot-id`, `--tag-specifications` | Valid AZ configured via `hive init` | Gateway validates input → NATS `ec2.CreateVolume` → daemon generates vol-ID via viperblock → creates empty volume of specified size → persists config.json to Predastore S3 → returns vol-ID with state=available | 1. Create 80GB gp3 volume
2. Boundary sizes (1 GiB min, 16384 GiB max)
3. Invalid AZ (error)
4. Verify volume in describe-volumes
5. Unsupported volume type (error - only gp3)
6. Size out of range (error) | **DONE** | | `delete-volume` | `--volume-id` | `--dry-run` | Volume must exist and be detached (state=available) | Gateway validates vol- prefix → NATS `ec2.DeleteVolume` → daemon confirms state=available and no AttachedInstance → NATS `ebs.delete` to viperblockd (stops nbdkit/WAL) → deletes S3 objects under vol-id/, vol-id-efi/, vol-id-cloudinit/ → returns success | 1. Delete detached volume
2. Delete attached volume (error: VolumeInUse)
3. Delete non-existent volume (error: InvalidVolume.NotFound)
4. Verify volume gone from describe-volumes
5. Malformed volume ID (error: InvalidVolumeID.Malformed)
6. Double delete (idempotent NotFound) | **DONE** | -| `attach-volume` | — | `--volume-id`, `--instance-id`, `--device` (e.g. /dev/sdf) | Volume must exist (available), instance must exist (running) | NATS to node running instance → start nbdkit for volume → QMP `blockdev-add` (NBD socket) → QMP `device_add` (virtio-blk-pci) → update instance metadata with attachment | 1. Attach volume to running instance
2. Verify device visible in guest OS
3. Attach already-attached volume (error)
4. Attach to non-existent instance (error)
5. Volume persists across stop/start | **NOT STARTED** | +| `attach-volume` | `--volume-id`, `--instance-id`, `--device` (optional, auto-assigns `/dev/sd[f-p]`) | `--dry-run` | Volume must exist (available), instance must exist (running) | Gateway sends to `ec2.cmd.{instanceId}` → daemon validates volume (Predastore) → `ebs.mount` via NATS (viperblock starts NBD server) → QMP `blockdev-add` (nbd-{volId}) → QMP `device_add` (virtio-blk-pci, vdisk-{volId}) → three-phase rollback on failure → update EBSRequests + BlockDeviceMappings → persist to JetStream + Predastore → respond with VolumeAttachment | 1. Attach volume to running instance
2. Auto-assign device name
3. Attach already-attached volume (VolumeInUse)
4. Attach to non-existent instance (InvalidInstanceID.NotFound)
5. Attach to stopped instance (IncorrectInstanceState)
6. Volume not found (InvalidVolume.NotFound)
7. All device slots full (AttachmentLimitExceeded)
8. Volume persists across stop/start | **DONE** | | `detach-volume` | — | `--volume-id`, `--instance-id`, `--device`, `--force` | Volume must be attached | NATS to node running instance → QMP `device_del` → QMP `blockdev-del` → terminate nbdkit for volume → sync WAL to Predastore → update instance metadata | 1. Detach from running instance
2. Force detach
3. Detach already-detached volume (error)
4. Verify device removed from guest OS | **NOT STARTED** | | `describe-volume-status` | — | `--volume-ids`, `--filters`, `--max-results` | None | Query viperblock for volume health status → return io-enabled status, availability zone, events | 1. Check healthy volume status
2. Check volume during modification
3. Filter by volume ID | **NOT STARTED** | | `describe-volumes-modifications` | — | `--volume-ids`, `--filters`, `--max-results` | None | Query pending/completed volume modifications → return modification state, progress, original/target size | 1. Check in-progress modification
2. Check completed modification
3. No modifications returns empty | **NOT STARTED** | diff --git a/docs/development/feature/attach-volume.md b/docs/development/feature/attach-volume.md new file mode 100644 index 000000000..d5aea0b3c --- /dev/null +++ b/docs/development/feature/attach-volume.md @@ -0,0 +1,125 @@ +# EC2 AttachVolume + +Hot-plug an EBS volume to a running instance via QMP, implementing the AWS `ec2 attach-volume` API. + +## Request Flow + +``` +AWS SDK (attach-volume) + -> Gateway (port 9999): parse AWS query params, validate input + -> NATS ec2.cmd.{instanceId}: per-instance topic routing + -> Daemon handleEC2Events: owns the instance's QEMU process + 1. Validate volume is "available" (Predastore/S3 lookup) + 2. ebs.mount via NATS -> viperblock starts NBD server + 3. QMP blockdev-add -> register NBD block device in QEMU + 4. QMP device_add -> attach as virtio-blk-pci to guest + 5. Update instance metadata + volume metadata + <- NATS response: ec2.VolumeAttachment JSON + <- Gateway: wrap in XML + <- AWS SDK: VolumeAttachment response +``` + +## Why Per-Instance NATS Routing + +AttachVolume uses `ec2.cmd.{instanceId}` instead of a generic `ec2.AttachVolume` topic. This is because the command must reach the specific daemon process that owns the QEMU instance. Each daemon subscribes to `ec2.cmd.{instanceId}` for every instance it manages. The same pattern is used by Start, Stop, and Terminate. If no daemon owns the instance, NATS returns `ErrNoResponders`, which the gateway maps to `InvalidInstanceID.NotFound`. + +Other volume operations (CreateVolume, DeleteVolume, DescribeVolumes) use generic topics like `ec2.CreateVolume` because they don't require access to a specific QEMU process. + +## Files Changed + +### Wire Format: `hive/qmp/qmp.go` + +Added `AttachVolume bool` to `Attributes` and a new `AttachVolumeData` struct carried as an optional pointer on `Command`. The pointer-with-omitempty pattern means the field is absent from JSON for non-attach commands, keeping the wire format clean. The `Attributes` struct uses boolean flags for dispatch (matching the existing Start/Stop/Terminate pattern). + +### Gateway Handler: `hive/gateway/ec2/volume/AttachVolume.go` + +Validates `VolumeId` and `InstanceId` are present, constructs a `qmp.Command` with `AttachVolume: true` and `AttachVolumeData`, sends it to `ec2.cmd.{instanceId}` via NATS request-response (30s timeout). The `Device` field is optional per the AWS API spec. + +NATS errors are differentiated: `ErrNoResponders` returns `InvalidInstanceID.NotFound` (no daemon owns this instance), while timeouts and connection errors return `InternalError` to avoid misleading users about non-existent instances when the real problem is infrastructure. + +### Gateway Switch: `hive/gateway/ec2.go` + +Added `"AttachVolume"` case to the EC2 action switch statement. + +### Daemon Handler: `hive/daemon/daemon_handlers.go` + +The core logic, placed as the first branch in `handleEC2Events` (before Start/Stop/Terminate). Implements a three-phase commit with reverse-order rollback: + +**Phase 1 - EBS Mount:** Sends `ebs.mount` via NATS to viperblock, which starts an NBD server and returns a URI (unix socket or TCP). If this fails, no cleanup is needed since nothing was mounted. + +**Phase 2 - QMP blockdev-add:** Registers the NBD block device in QEMU. Uses `nbd-{volumeId}` as the node-name for predictable identification during future detach. The `export` field is set to `""` per viperblock convention (default export). If this fails, Phase 1 is rolled back via `ebs.unmount`. + +**Phase 3 - QMP device_add:** Attaches the block device to the guest as `virtio-blk-pci` with id `vdisk-{volumeId}`. If this fails, Phase 2 is rolled back via `blockdev-del`, then Phase 1 via `ebs.unmount`. The `blockdev-del` rollback error is checked: if it fails, the EBS unmount is skipped to avoid tearing down storage that QEMU still references (which could crash the VM). + +After all three phases succeed, the handler updates in-memory state (`EBSRequests`, `BlockDeviceMappings`), persists to Predastore (volume metadata) and JetStream (instance state), then responds with `ec2.VolumeAttachment` JSON. + +### Volume Service: `hive/handlers/ec2/volume/service_impl.go` + +`getVolumeConfig` was renamed to `GetVolumeConfig` (exported directly instead of using a wrapper) so the daemon can validate that a volume exists and check its state before attempting attachment. + +`UpdateVolumeState` was extended with a `deviceName` parameter rather than creating a separate `UpdateVolumeAttachment` method. The existing callers pass `""` for deviceName since boot volumes don't track device names this way. This avoids near-duplicate methods. + +### Volume Service Interface: `hive/handlers/ec2/volume/service.go` + +The `VolumeService` interface is kept to AWS-facing operations only (CreateVolume, DescribeVolumes, ModifyVolume, DeleteVolume). `GetVolumeConfig` and `UpdateVolumeState` are methods on the concrete `VolumeServiceImpl` type, not the interface, because they are internal operations used by the daemon rather than AWS API endpoints. The daemon holds a `*VolumeServiceImpl` directly so it can call these methods without polluting the interface that the gateway-side `NATSVolumeService` also implements. + +### Device Name Assignment: `hive/daemon/daemon.go` + +`nextAvailableDevice` scans both `EBSRequests` and `BlockDeviceMappings` to find the next available `/dev/sd[f-p]` slot. Both collections are checked because there's a window during attachment where a volume appears in `EBSRequests` but not yet in `BlockDeviceMappings`. Returns `""` when all 11 slots are exhausted, which the handler maps to `AttachmentLimitExceeded`. + +The `/dev/sd[f-p]` range follows the AWS convention. The guest OS sees these as `/dev/vd*` when using virtio drivers, but the metadata device name is the `/dev/sd*` form per AWS API compatibility. + +### Rollback Helper: `hive/daemon/daemon.go` + +`rollbackEBSMount` sends `ebs.unmount` via NATS and verifies the response: checks both the `Error` field and the `Mounted` boolean. Rollback failures are logged at Error level but not propagated to callers since rollback is best-effort cleanup during an already-failed operation. This is a deliberate tradeoff: propagating rollback errors would complicate the handler without changing the user-visible outcome (the attach already failed). + +### Config: `hive/config/config.go` + +Added `DeviceName` field to `EBSRequest`. This tracks which `/dev/sd*` device name is assigned to a hot-plugged volume. Boot-time volumes don't use this field since their device assignment is handled by QEMU command-line args. + +### NBD URI Parsing: `hive/utils/utils.go` + +`ParseNBDURI` is the inverse of the existing `FormatNBDSocketURI`/`FormatNBDTCPURI` functions. It extracts the components needed for QMP's `blockdev-add` server argument: +- `nbd:unix:/path/to/socket.sock` -> `type=unix, path=/path/to/socket.sock` +- `nbd://host:port` -> `type=inet, host=host, port=port` + +## Design Decisions + +**Why QMP hot-plug instead of restarting the VM?** Hot-plug via `blockdev-add` + `device_add` attaches the volume without downtime. This matches AWS behavior where `attach-volume` works on running instances immediately. + +**Why `nbd-{volumeId}` and `vdisk-{volumeId}` naming?** These are predictable identifiers derived from the volume ID. This makes future `detach-volume` straightforward: the daemon can construct the node-name and device-id from the volume ID without maintaining a separate mapping. + +**Why `export: ""`?** Viperblock's NBD server uses a blank export name as the default. This is a viperblock convention, not a universal NBD convention. + +**Why rollback skips EBS unmount when blockdev-del fails?** If `blockdev-del` fails, QEMU still has a reference to the NBD backend. Unmounting the EBS volume would tear down the NBD server while QEMU still tries to do I/O through it, potentially crashing or hanging the VM. Leaving the mount in place is the safer failure mode: the volume is leaked but the VM stays healthy. + +**Why `DeleteOnTermination` defaults to false?** This matches AWS behavior: hot-attached volumes default to `DeleteOnTermination: false`. Boot volumes set this to true. Users can change it via `modify-instance-attribute`. + +## Error Codes + +| Condition | AWS Error Code | +|-----------|---------------| +| Missing VolumeId/InstanceId | `InvalidParameterValue` | +| Volume not found in Predastore | `InvalidVolume.NotFound` | +| Volume already in-use | `VolumeInUse` | +| Instance not found (no NATS responder) | `InvalidInstanceID.NotFound` | +| Instance not running | `IncorrectInstanceState` | +| All device slots full | `AttachmentLimitExceeded` | +| Internal failures (mount, QMP, etc.) | `InternalError` | + +## Testing + +**Gateway validation** (`AttachVolume_test.go`): Table-driven tests for nil input, empty input, missing fields, empty strings, valid inputs with/without device. Follows the pattern from `CreateVolume_test.go`. + +**Device name assignment** (`daemon_test.go`): 6 scenarios covering empty instance, existing block device mappings, existing EBS requests, all devices used, nil instance, mixed sources. + +**Handler error paths** (`daemon_test.go`): Tests for missing attach volume data, instance not running, and volume not found. Uses a real embedded NATS server (not mocks). The happy path is not currently testable because `volumeService` is a concrete type — testing the full flow would require introducing interface-based dependency injection for the volume service and QMP client. + +**NBD URI parsing** (`utils_test.go`): 8 cases covering both formats, empty paths, missing ports, invalid ports, unsupported formats. + +## Known Limitations + +- No validation of user-supplied device names against the `/dev/sd[f-p]` range or conflict checking. +- TOCTOU gap between checking instance status/volume state and performing the actual QMP operations. A concurrent stop could race with an attach. +- The `Attributes` struct uses boolean flags for mutually exclusive operations. A string-based action field would be structurally safer but is a breaking wire-format change. +- Volume metadata and JetStream state persistence failures after a successful attach are logged but do not cause the response to indicate failure, since the volume is already physically attached to the VM. diff --git a/docs/development/feature/terminate-on-delete.md b/docs/development/features/terminate-on-delete.md similarity index 100% rename from docs/development/feature/terminate-on-delete.md rename to docs/development/features/terminate-on-delete.md diff --git a/go.mod b/go.mod index 45f1635c8..031678333 100644 --- a/go.mod +++ b/go.mod @@ -9,7 +9,7 @@ require ( github.com/google/uuid v1.6.0 github.com/kdomanski/iso9660 v0.4.0 github.com/mulgadc/predastore v1.0.0 - github.com/mulgadc/viperblock v1.0.4 + github.com/mulgadc/viperblock v1.0.5 github.com/nats-io/nats-server/v2 v2.11.6 github.com/nats-io/nats.go v1.43.0 github.com/pelletier/go-toml/v2 v2.2.4 diff --git a/go.sum b/go.sum index b6157c2f5..ce854a44f 100644 --- a/go.sum +++ b/go.sum @@ -276,6 +276,8 @@ github.com/mulgadc/predastore v1.0.0 h1:IZ5V/EL+k+zz0zGoTYp+1BgODb43JxHsb/kvORuy github.com/mulgadc/predastore v1.0.0/go.mod h1:x46A1z091hlZkMt3kkcuLSy2PAzSgROr6ft04qcYOU4= github.com/mulgadc/viperblock v1.0.4 h1:fH4jqwnnvWXcEpRk73oiJPyVuTCuufWaV2mO0R91h7o= github.com/mulgadc/viperblock v1.0.4/go.mod h1:CuoSKPcR50M11DmadqXYAeOo5HNI6Ix5B/OvSzuLWxA= +github.com/mulgadc/viperblock v1.0.5 h1:TWLWWWHT9r25CEMHN7UxLDc6QGsi8T6l7hlTHmJb4nk= +github.com/mulgadc/viperblock v1.0.5/go.mod h1:CuoSKPcR50M11DmadqXYAeOo5HNI6Ix5B/OvSzuLWxA= github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= github.com/nats-io/jwt/v2 v2.7.4 h1:jXFuDDxs/GQjGDZGhNgH4tXzSUK6WQi2rsj4xmsNOtI= diff --git a/hive/config/config.go b/hive/config/config.go index 626aa233b..a0743144c 100644 --- a/hive/config/config.go +++ b/hive/config/config.go @@ -97,6 +97,7 @@ type EBSRequest struct { CloudInit bool DeleteOnTermination bool NBDURI string // NBD URI - socket path (nbd:unix:/path.sock) or TCP (nbd://host:port) + DeviceName string // Device name (e.g. /dev/sdf) for hot-plugged volumes } // NBDTransport defines the transport type for NBD connections diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index b5ed07db8..bed9685d5 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -985,9 +985,9 @@ func (d *Daemon) stopInstance(instances map[string]*vm.VM, deleteVolume bool) er slog.Info("Unmounted Viperblock volume", "id", instance.ID, "data", string(msg.Data)) } - // Update volume state to "available" for boot volumes - if ebsRequest.Boot { - if err := d.volumeService.UpdateVolumeState(ebsRequest.Name, "available", ""); err != nil { + // Update volume state to "available" for all user-visible volumes (boot + hot-attached) + if !ebsRequest.EFI && !ebsRequest.CloudInit { + if err := d.volumeService.UpdateVolumeState(ebsRequest.Name, "available", "", ""); err != nil { slog.Error("Failed to update volume state to available", "volumeId", ebsRequest.Name, "err", err) } } @@ -1233,7 +1233,7 @@ func (d *Daemon) LaunchInstance(instance *vm.VM) (err error) { instance.EBSRequests.Mu.Lock() for _, ebsReq := range instance.EBSRequests.Requests { if ebsReq.Boot { - if err := d.volumeService.UpdateVolumeState(ebsReq.Name, "in-use", instance.ID); err != nil { + if err := d.volumeService.UpdateVolumeState(ebsReq.Name, "in-use", instance.ID, ""); err != nil { slog.Error("Failed to update volume state to in-use", "volumeId", ebsReq.Name, "err", err) } } @@ -1302,6 +1302,14 @@ func (d *Daemon) StartInstance(instance *vm.VM) error { Architecture: architecture, } + // Add PCIe root ports for volume hotplug (Q35 requires explicit root ports). + // 11 ports for /dev/sd[f-p] hotplug slots, starting at chassis 1. + for i := 1; i <= 11; i++ { + instance.Config.Devices = append(instance.Config.Devices, vm.Device{ + Value: fmt.Sprintf("pcie-root-port,id=hotplug%d,chassis=%d,slot=0", i, i), + }) + } + // Loop through each volume in volumes instance.EBSRequests.Mu.Lock() @@ -1575,6 +1583,69 @@ func (d *Daemon) MountVolumes(instance *vm.VM) error { } +// 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 config.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("ebs.unmount", data, 10*time.Second) + if err != nil { + slog.Error("rollbackEBSMount: ebs.unmount NATS request failed", "volume", req.Name, "err", err) + return + } + var resp config.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) +} + +// 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 { diff --git a/hive/daemon/daemon_handlers.go b/hive/daemon/daemon_handlers.go index 30c7e3c58..746c4f2bb 100644 --- a/hive/daemon/daemon_handlers.go +++ b/hive/daemon/daemon_handlers.go @@ -76,7 +76,7 @@ func (d *Daemon) handleEC2StartInstances(msg *nats.Msg) { } -// handleEC2Events processes incoming EC2 events (start, stop, terminate) +// handleEC2Events processes incoming EC2 instance events (start, stop, terminate, attach-volume) func (d *Daemon) handleEC2Events(msg *nats.Msg) { var command qmp.Command @@ -106,6 +106,236 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) { return } + // AttachVolume 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) + if command.Attributes.AttachVolume { + 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(awserrors.ErrorInvalidParameterValue) + return + } + + volumeID := command.AttachVolumeData.VolumeID + device := command.AttachVolumeData.Device + + // Validate instance is running + d.Instances.Mu.Lock() + status := instance.Status + d.Instances.Mu.Unlock() + + if status != vm.StateRunning { + slog.Error("AttachVolume: instance not running", "instanceId", command.ID, "status", status) + respondWithError(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) + respondWithError(awserrors.ErrorInvalidVolumeNotFound) + return + } + + if volCfg.VolumeMetadata.State != "available" { + slog.Error("AttachVolume: volume not available", "volumeId", volumeID, "state", volCfg.VolumeMetadata.State) + respondWithError(awserrors.ErrorVolumeInUse) + return + } + + // Determine device name + if device == "" { + d.Instances.Mu.Lock() + device = nextAvailableDevice(instance) + d.Instances.Mu.Unlock() + if device == "" { + slog.Error("AttachVolume: no available device names") + respondWithError(awserrors.ErrorAttachmentLimitExceeded) + return + } + } + + // Create EBS mount request + ebsRequest := config.EBSRequest{ + Name: volumeID, + DeviceName: device, + } + + ebsMountData, err := json.Marshal(ebsRequest) + if err != nil { + slog.Error("AttachVolume: failed to marshal ebs.mount request", "err", err) + respondWithError(awserrors.ErrorServerInternal) + return + } + + mountReply, err := d.natsConn.Request("ebs.mount", ebsMountData, 30*time.Second) + if err != nil { + slog.Error("AttachVolume: ebs.mount failed", "volumeId", volumeID, "err", err) + respondWithError(awserrors.ErrorServerInternal) + return + } + + var mountResp config.EBSMountResponse + if err := json.Unmarshal(mountReply.Data, &mountResp); err != nil { + slog.Error("AttachVolume: failed to unmarshal mount response", "err", err) + respondWithError(awserrors.ErrorServerInternal) + return + } + + if mountResp.Error != "" { + slog.Error("AttachVolume: mount returned error", "volumeId", volumeID, "err", mountResp.Error) + respondWithError(awserrors.ErrorServerInternal) + return + } + + nbdURI := mountResp.URI + if nbdURI == "" { + slog.Error("AttachVolume: mount response has empty URI", "volumeId", volumeID) + d.rollbackEBSMount(ebsRequest) + respondWithError(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(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": fmt.Sprintf("%d", nbdPort)} + } + + nodeName := fmt.Sprintf("nbd-%s", volumeID) + deviceID := fmt.Sprintf("vdisk-%s", volumeID) + + // 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(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, + } + 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(awserrors.ErrorServerInternal) + return + } + + // Update instance state: append EBSRequest + instance.EBSRequests.Mu.Lock() + instance.EBSRequests.Requests = append(instance.EBSRequests.Requests, ebsRequest) + instance.EBSRequests.Mu.Unlock() + + // Update BlockDeviceMappings on the ec2.Instance + d.Instances.Mu.Lock() + if instance.Instance != nil { + now := time.Now() + mapping := &ec2.InstanceBlockDeviceMapping{} + mapping.SetDeviceName(device) + mapping.Ebs = &ec2.EbsInstanceBlockDevice{} + mapping.Ebs.SetVolumeId(volumeID) + mapping.Ebs.SetAttachTime(now) + mapping.Ebs.SetDeleteOnTermination(false) + mapping.Ebs.SetStatus("attached") + instance.Instance.BlockDeviceMappings = append(instance.Instance.BlockDeviceMappings, mapping) + } + d.Instances.Mu.Unlock() + + // Update volume metadata in S3 + if err := d.volumeService.UpdateVolumeState(volumeID, "in-use", command.ID, device); 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) + } + + // Respond with VolumeAttachment + now := time.Now() + attachment := ec2.VolumeAttachment{ + VolumeId: aws.String(volumeID), + InstanceId: aws.String(command.ID), + Device: aws.String(device), + State: aws.String("attached"), + AttachTime: aws.Time(now), + DeleteOnTermination: aws.Bool(false), + } + + jsonResp, err := json.Marshal(attachment) + if err != nil { + slog.Error("AttachVolume: failed to marshal response", "err", err) + respondWithError(awserrors.ErrorServerInternal) + return + } + + msg.Respond(jsonResp) + slog.Info("Volume attached successfully", "volumeId", volumeID, "instanceId", command.ID, "device", device) + return + } + // Start an instance if command.Attributes.StartInstance { slog.Info("Starting instance", "id", command.ID) diff --git a/hive/daemon/daemon_test.go b/hive/daemon/daemon_test.go index 1fe558473..390622b77 100644 --- a/hive/daemon/daemon_test.go +++ b/hive/daemon/daemon_test.go @@ -1935,6 +1935,197 @@ func TestStopInstance_NoDelete_OnStop(t *testing.T) { assert.Empty(t, ebsDeletedVolumes, "No volumes should be deleted during stop (not terminate)") } +// TestNextAvailableDevice tests the device name auto-assignment logic +func TestNextAvailableDevice(t *testing.T) { + t.Run("EmptyInstance_ReturnsFirstDevice", func(t *testing.T) { + instance := &vm.VM{ + ID: "i-test", + Instance: &ec2.Instance{}, + } + dev := nextAvailableDevice(instance) + assert.Equal(t, "/dev/sdf", dev) + }) + + t.Run("WithExistingBlockDeviceMappings", func(t *testing.T) { + instance := &vm.VM{ + ID: "i-test", + Instance: &ec2.Instance{ + BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{ + {DeviceName: aws.String("/dev/sdf")}, + {DeviceName: aws.String("/dev/sdg")}, + }, + }, + } + dev := nextAvailableDevice(instance) + assert.Equal(t, "/dev/sdh", dev) + }) + + t.Run("WithExistingEBSRequests", func(t *testing.T) { + instance := &vm.VM{ + ID: "i-test", + Instance: &ec2.Instance{}, + EBSRequests: config.EBSRequests{ + Requests: []config.EBSRequest{ + {Name: "vol-1", DeviceName: "/dev/sdf"}, + }, + }, + } + dev := nextAvailableDevice(instance) + assert.Equal(t, "/dev/sdg", dev) + }) + + t.Run("AllDevicesUsed_ReturnsEmpty", func(t *testing.T) { + bdms := make([]*ec2.InstanceBlockDeviceMapping, 0) + for c := 'f'; c <= 'p'; c++ { + dev := fmt.Sprintf("/dev/sd%c", c) + bdms = append(bdms, &ec2.InstanceBlockDeviceMapping{ + DeviceName: aws.String(dev), + }) + } + instance := &vm.VM{ + ID: "i-test", + Instance: &ec2.Instance{ + BlockDeviceMappings: bdms, + }, + } + dev := nextAvailableDevice(instance) + assert.Equal(t, "", dev) + }) + + t.Run("NilInstance_ReturnsFirstDevice", func(t *testing.T) { + instance := &vm.VM{ + ID: "i-test", + } + dev := nextAvailableDevice(instance) + assert.Equal(t, "/dev/sdf", dev) + }) + + t.Run("MixedSources_SkipsAll", func(t *testing.T) { + instance := &vm.VM{ + ID: "i-test", + Instance: &ec2.Instance{ + BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{ + {DeviceName: aws.String("/dev/sdf")}, + }, + }, + EBSRequests: config.EBSRequests{ + Requests: []config.EBSRequest{ + {Name: "vol-1", DeviceName: "/dev/sdg"}, + }, + }, + } + dev := nextAvailableDevice(instance) + assert.Equal(t, "/dev/sdh", dev) + }) +} + +// TestHandleEC2Events_AttachVolume tests the attach-volume handler in handleEC2Events +func TestHandleEC2Events_AttachVolume(t *testing.T) { + ns, natsURL := startTestNATSServer(t) + defer ns.Shutdown() + + daemon := createTestDaemon(t, natsURL) + + instanceID := "i-test-attach" + volumeID := "vol-test-attach" + instanceType := getTestInstanceType() + + // Create a running instance (no actual QMP client - will fail at QMP step) + instance := &vm.VM{ + ID: instanceID, + InstanceType: instanceType, + Status: vm.StateRunning, + Instance: &ec2.Instance{}, + QMPClient: &qmp.QMPClient{}, // nil encoder/decoder + } + daemon.Instances.VMS[instanceID] = instance + + // Subscribe the handler to the instance's per-instance topic + sub, err := daemon.natsConn.QueueSubscribe( + fmt.Sprintf("ec2.cmd.%s", instanceID), + "hive-events", + daemon.handleEC2Events, + ) + require.NoError(t, err) + defer sub.Unsubscribe() + + t.Run("MissingAttachVolumeData", func(t *testing.T) { + command := qmp.Command{ + ID: instanceID, + Attributes: qmp.Attributes{ + AttachVolume: true, + }, + // No AttachVolumeData + } + data, _ := json.Marshal(command) + + resp, err := daemon.natsConn.Request( + fmt.Sprintf("ec2.cmd.%s", instanceID), + data, + 5*time.Second, + ) + require.NoError(t, err) + + // Should return an error payload + assert.Contains(t, string(resp.Data), "InvalidParameterValue") + }) + + t.Run("InstanceNotRunning", func(t *testing.T) { + // Temporarily set status to stopped + daemon.Instances.Mu.Lock() + instance.Status = vm.StateStopped + daemon.Instances.Mu.Unlock() + + command := qmp.Command{ + ID: instanceID, + Attributes: qmp.Attributes{ + AttachVolume: true, + }, + AttachVolumeData: &qmp.AttachVolumeData{ + VolumeID: volumeID, + }, + } + data, _ := json.Marshal(command) + + resp, err := daemon.natsConn.Request( + fmt.Sprintf("ec2.cmd.%s", instanceID), + data, + 5*time.Second, + ) + require.NoError(t, err) + assert.Contains(t, string(resp.Data), "IncorrectInstanceState") + + // Restore running state + daemon.Instances.Mu.Lock() + instance.Status = vm.StateRunning + daemon.Instances.Mu.Unlock() + }) + + t.Run("VolumeNotFound", func(t *testing.T) { + // volumeService.GetVolumeConfig will fail since we have no S3 backend + command := qmp.Command{ + ID: instanceID, + Attributes: qmp.Attributes{ + AttachVolume: true, + }, + AttachVolumeData: &qmp.AttachVolumeData{ + VolumeID: "vol-nonexistent", + Device: "/dev/sdf", + }, + } + data, _ := json.Marshal(command) + + resp, err := daemon.natsConn.Request( + fmt.Sprintf("ec2.cmd.%s", instanceID), + data, + 5*time.Second, + ) + require.NoError(t, err) + // Should fail at volume validation (no S3 backend) + assert.Contains(t, string(resp.Data), "InvalidVolume.NotFound") + }) +} + // TestEBSRequest_JSON_Serialization verifies DeleteOnTermination survives JSON round-trip // (important for JetStream state persistence) func TestEBSRequest_JSON_Serialization(t *testing.T) { diff --git a/hive/gateway/ec2.go b/hive/gateway/ec2.go index 348d5de5a..57385d864 100644 --- a/hive/gateway/ec2.go +++ b/hive/gateway/ec2.go @@ -393,6 +393,27 @@ func (gw *GatewayConfig) EC2_Request(ctx *fiber.Ctx) error { return errors.New("failed to marshal response to XML") } + case "AttachVolume": + var input = &ec2.AttachVolumeInput{} + err = awsec2query.QueryParamsToStruct(queryArgs, input) + + if err != nil { + return err + } + + output, err := gateway_ec2_volume.AttachVolume(input, gw.NATSConn) + + if err != nil { + return err + } + + payload := utils.GenerateXMLPayload("AttachVolumeResponse", output) + xmlOutput, err = utils.MarshalToXML(payload) + + if err != nil { + return errors.New("failed to marshal response to XML") + } + default: err = errors.New("InvalidAction") } diff --git a/hive/gateway/ec2/volume/AttachVolume.go b/hive/gateway/ec2/volume/AttachVolume.go new file mode 100644 index 000000000..40672f8bc --- /dev/null +++ b/hive/gateway/ec2/volume/AttachVolume.go @@ -0,0 +1,89 @@ +package gateway_ec2_volume + +import ( + "encoding/json" + "errors" + "fmt" + "log/slog" + "time" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/hive/hive/awserrors" + "github.com/mulgadc/hive/hive/qmp" + "github.com/mulgadc/hive/hive/utils" + "github.com/nats-io/nats.go" +) + +// ValidateAttachVolumeInput validates the input parameters for AttachVolume +func ValidateAttachVolumeInput(input *ec2.AttachVolumeInput) error { + if input == nil { + return errors.New(awserrors.ErrorInvalidParameterValue) + } + + if input.VolumeId == nil || *input.VolumeId == "" { + return errors.New(awserrors.ErrorInvalidParameterValue) + } + + if input.InstanceId == nil || *input.InstanceId == "" { + return errors.New(awserrors.ErrorInvalidParameterValue) + } + + return nil +} + +// AttachVolume sends an attach-volume command to the daemon owning the instance +func AttachVolume(input *ec2.AttachVolumeInput, natsConn *nats.Conn) (ec2.VolumeAttachment, error) { + var output ec2.VolumeAttachment + + if err := ValidateAttachVolumeInput(input); err != nil { + return output, err + } + + instanceID := *input.InstanceId + volumeID := *input.VolumeId + + device := "" + if input.Device != nil { + device = *input.Device + } + + command := qmp.Command{ + ID: instanceID, + Attributes: qmp.Attributes{ + AttachVolume: true, + }, + AttachVolumeData: &qmp.AttachVolumeData{ + VolumeID: volumeID, + Device: device, + }, + } + + jsonData, err := json.Marshal(command) + if err != nil { + slog.Error("AttachVolume: Failed to marshal command", "err", err) + return output, errors.New(awserrors.ErrorServerInternal) + } + + subject := fmt.Sprintf("ec2.cmd.%s", instanceID) + msg, err := natsConn.Request(subject, jsonData, 30*time.Second) + if err != nil { + slog.Error("AttachVolume: NATS request failed", "instanceId", instanceID, "volumeId", volumeID, "err", err) + if errors.Is(err, nats.ErrNoResponders) { + return output, errors.New(awserrors.ErrorInvalidInstanceIDNotFound) + } + return output, errors.New(awserrors.ErrorServerInternal) + } + + responseError, err := utils.ValidateErrorPayload(msg.Data) + if err != nil { + return output, errors.New(*responseError.Code) + } + + if err := json.Unmarshal(msg.Data, &output); err != nil { + slog.Error("AttachVolume: Failed to unmarshal response", "err", err) + return output, errors.New(awserrors.ErrorServerInternal) + } + + slog.Info("AttachVolume completed", "instanceId", instanceID, "volumeId", volumeID) + return output, nil +} diff --git a/hive/gateway/ec2/volume/AttachVolume_test.go b/hive/gateway/ec2/volume/AttachVolume_test.go new file mode 100644 index 000000000..9a6b471f6 --- /dev/null +++ b/hive/gateway/ec2/volume/AttachVolume_test.go @@ -0,0 +1,96 @@ +package gateway_ec2_volume + +import ( + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/hive/hive/awserrors" + "github.com/stretchr/testify/assert" +) + +func TestValidateAttachVolumeInput(t *testing.T) { + tests := []struct { + name string + input *ec2.AttachVolumeInput + wantErr bool + errMsg string + }{ + { + name: "NilInput", + input: nil, + wantErr: true, + errMsg: awserrors.ErrorInvalidParameterValue, + }, + { + name: "EmptyInput", + input: &ec2.AttachVolumeInput{}, + wantErr: true, + errMsg: awserrors.ErrorInvalidParameterValue, + }, + { + name: "MissingVolumeId", + input: &ec2.AttachVolumeInput{ + InstanceId: aws.String("i-1234567890abcdef0"), + }, + wantErr: true, + errMsg: awserrors.ErrorInvalidParameterValue, + }, + { + name: "MissingInstanceId", + input: &ec2.AttachVolumeInput{ + VolumeId: aws.String("vol-1234567890abcdef0"), + }, + wantErr: true, + errMsg: awserrors.ErrorInvalidParameterValue, + }, + { + name: "EmptyVolumeId", + input: &ec2.AttachVolumeInput{ + VolumeId: aws.String(""), + InstanceId: aws.String("i-1234567890abcdef0"), + }, + wantErr: true, + errMsg: awserrors.ErrorInvalidParameterValue, + }, + { + name: "EmptyInstanceId", + input: &ec2.AttachVolumeInput{ + VolumeId: aws.String("vol-1234567890abcdef0"), + InstanceId: aws.String(""), + }, + wantErr: true, + errMsg: awserrors.ErrorInvalidParameterValue, + }, + { + name: "ValidInput_NoDevice", + input: &ec2.AttachVolumeInput{ + VolumeId: aws.String("vol-1234567890abcdef0"), + InstanceId: aws.String("i-1234567890abcdef0"), + }, + wantErr: false, + }, + { + name: "ValidInput_WithDevice", + input: &ec2.AttachVolumeInput{ + VolumeId: aws.String("vol-1234567890abcdef0"), + InstanceId: aws.String("i-1234567890abcdef0"), + Device: aws.String("/dev/sdf"), + }, + wantErr: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := ValidateAttachVolumeInput(tt.input) + + if tt.wantErr { + assert.Error(t, err) + assert.Equal(t, tt.errMsg, err.Error()) + } else { + assert.NoError(t, err) + } + }) + } +} diff --git a/hive/handlers/ec2/volume/service_impl.go b/hive/handlers/ec2/volume/service_impl.go index c59dda967..94adddf58 100644 --- a/hive/handlers/ec2/volume/service_impl.go +++ b/hive/handlers/ec2/volume/service_impl.go @@ -287,7 +287,7 @@ func (s *VolumeServiceImpl) fetchVolumesByIDs(volumeIDs []*string) []*ec2.Volume // getVolumeByID fetches a single volume's config from S3 and builds an EC2 Volume func (s *VolumeServiceImpl) getVolumeByID(volumeID string) (*ec2.Volume, error) { - cfg, err := s.getVolumeConfig(volumeID) + cfg, err := s.GetVolumeConfig(volumeID) if err != nil { return nil, err } @@ -362,8 +362,8 @@ type volumeConfigWrapper struct { VolumeConfig viperblock.VolumeConfig `json:"VolumeConfig"` } -// getVolumeConfig reads the raw VolumeConfig from S3 for a given volume ID -func (s *VolumeServiceImpl) getVolumeConfig(volumeID string) (*viperblock.VolumeConfig, error) { +// GetVolumeConfig reads the raw VolumeConfig from S3 for a given volume ID. +func (s *VolumeServiceImpl) GetVolumeConfig(volumeID string) (*viperblock.VolumeConfig, error) { configKey := volumeID + "/config.json" getResult, err := s.s3Client.GetObject(&s3.GetObjectInput{ @@ -455,15 +455,16 @@ func (s *VolumeServiceImpl) mergeVolumeConfig(configKey string, cfg *viperblock. return json.Marshal(state) } -// UpdateVolumeState updates the State and AttachedInstance fields for a volume in S3. -func (s *VolumeServiceImpl) UpdateVolumeState(volumeID, state, attachedInstance string) error { - cfg, err := s.getVolumeConfig(volumeID) +// UpdateVolumeState updates volume metadata (state, attachment, device) in the object store. +func (s *VolumeServiceImpl) UpdateVolumeState(volumeID, state, attachedInstance, deviceName string) error { + cfg, err := s.GetVolumeConfig(volumeID) if err != nil { return fmt.Errorf("failed to get volume config for state update: %w", err) } cfg.VolumeMetadata.State = state cfg.VolumeMetadata.AttachedInstance = attachedInstance + cfg.VolumeMetadata.DeviceName = deviceName if attachedInstance != "" { cfg.VolumeMetadata.AttachedAt = time.Now() } @@ -472,7 +473,7 @@ func (s *VolumeServiceImpl) UpdateVolumeState(volumeID, state, attachedInstance return fmt.Errorf("failed to write volume config for state update: %w", err) } - slog.Info("Updated volume state", "volumeId", volumeID, "state", state, "attachedInstance", attachedInstance) + slog.Info("Updated volume state", "volumeId", volumeID, "state", state, "attachedInstance", attachedInstance, "deviceName", deviceName) return nil } @@ -485,7 +486,7 @@ func (s *VolumeServiceImpl) ModifyVolume(input *ec2.ModifyVolumeInput) (*ec2.Mod volumeID := *input.VolumeId slog.Info("ModifyVolume request", "volumeId", volumeID) - cfg, err := s.getVolumeConfig(volumeID) + cfg, err := s.GetVolumeConfig(volumeID) if err != nil { slog.Error("ModifyVolume failed to get volume config", "volumeId", volumeID, "err", err) return nil, err @@ -569,7 +570,7 @@ func (s *VolumeServiceImpl) DeleteVolume(input *ec2.DeleteVolumeInput) (*ec2.Del slog.Info("DeleteVolume request", "volumeId", volumeID) // Fetch volume config to validate state - cfg, err := s.getVolumeConfig(volumeID) + cfg, err := s.GetVolumeConfig(volumeID) if err != nil { slog.Error("DeleteVolume failed to get volume config", "volumeId", volumeID, "err", err) return nil, err diff --git a/hive/qmp/qmp.go b/hive/qmp/qmp.go index f56227682..1aba90041 100644 --- a/hive/qmp/qmp.go +++ b/hive/qmp/qmp.go @@ -21,15 +21,22 @@ var CommandResponseTypes = map[string]unmarshalTarget{ } type Command struct { - ID string `json:"id"` - QMPCommand QMPCommand `json:"command"` - Attributes Attributes `json:"attributes"` + ID string `json:"id"` + QMPCommand QMPCommand `json:"command"` + Attributes Attributes `json:"attributes"` + AttachVolumeData *AttachVolumeData `json:"attach_volume_data,omitempty"` } type Attributes struct { StopInstance bool `json:"stop_instance"` TerminateInstance bool `json:"delete_instance"` StartInstance bool `json:"start_instance"` + AttachVolume bool `json:"attach_volume"` +} + +type AttachVolumeData struct { + VolumeID string `json:"volume_id"` + Device string `json:"device,omitempty"` } // (QEMU) stop / cont (resume) / system_powerdown / system_reset / system_wakeup diff --git a/hive/utils/utils.go b/hive/utils/utils.go index 34abc5df1..1d96bbb15 100644 --- a/hive/utils/utils.go +++ b/hive/utils/utils.go @@ -282,6 +282,36 @@ func FormatNBDTCPURI(host string, port int) string { return fmt.Sprintf("nbd://%s:%d", host, port) } +// ParseNBDURI parses an NBD URI into its components for QMP blockdev-add. +// Supported formats: +// - "nbd:unix:/path/to/socket.sock" → serverType="unix", path="/path/to/socket.sock" +// - "nbd://host:port" → serverType="inet", host="host", port= +func ParseNBDURI(nbdURI string) (serverType, path, host string, port int, err error) { + if strings.HasPrefix(nbdURI, "nbd:unix:") { + path = strings.TrimPrefix(nbdURI, "nbd:unix:") + if path == "" { + return "", "", "", 0, fmt.Errorf("empty socket path in NBD URI: %s", nbdURI) + } + return "unix", path, "", 0, nil + } + + if strings.HasPrefix(nbdURI, "nbd://") { + hostPort := strings.TrimPrefix(nbdURI, "nbd://") + colonIdx := strings.LastIndex(hostPort, ":") + if colonIdx < 0 { + return "", "", "", 0, fmt.Errorf("missing port in NBD TCP URI: %s", nbdURI) + } + host = hostPort[:colonIdx] + port, err = strconv.Atoi(hostPort[colonIdx+1:]) + if err != nil { + return "", "", "", 0, fmt.Errorf("invalid port in NBD URI: %s", nbdURI) + } + return "inet", "", host, port, nil + } + + return "", "", "", 0, fmt.Errorf("unsupported NBD URI format: %s", nbdURI) +} + func WritePidFile(name string, pid int) error { // Write PID to file, check XDG, otherwise user home directory ~/hive/ diff --git a/hive/utils/utils_test.go b/hive/utils/utils_test.go index 2ee92a057..485520975 100644 --- a/hive/utils/utils_test.go +++ b/hive/utils/utils_test.go @@ -314,6 +314,81 @@ func TestValidateErrorPayload(t *testing.T) { } } +func TestParseNBDURI(t *testing.T) { + tests := []struct { + name string + uri string + wantType string + wantPath string + wantHost string + wantPort int + wantErr bool + }{ + { + name: "Unix socket", + uri: "nbd:unix:/run/user/1000/nbd-vol-123.sock", + wantType: "unix", + wantPath: "/run/user/1000/nbd-vol-123.sock", + }, + { + name: "TCP address", + uri: "nbd://127.0.0.1:34305", + wantType: "inet", + wantHost: "127.0.0.1", + wantPort: 34305, + }, + { + name: "TCP with hostname", + uri: "nbd://storage.local:9000", + wantType: "inet", + wantHost: "storage.local", + wantPort: 9000, + }, + { + name: "Empty socket path", + uri: "nbd:unix:", + wantErr: true, + }, + { + name: "Missing port in TCP", + uri: "nbd://127.0.0.1", + wantErr: true, + }, + { + name: "Invalid port", + uri: "nbd://127.0.0.1:notaport", + wantErr: true, + }, + { + name: "Unsupported format", + uri: "http://example.com", + wantErr: true, + }, + { + name: "Empty string", + uri: "", + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + serverType, path, host, port, err := ParseNBDURI(tt.uri) + + if tt.wantErr { + assert.Error(t, err) + return + } + + assert.NoError(t, err) + assert.Equal(t, tt.wantType, serverType) + assert.Equal(t, tt.wantPath, path) + assert.Equal(t, tt.wantHost, host) + assert.Equal(t, tt.wantPort, port) + }) + } +} + func TestMarshalToXML(t *testing.T) { type TestStruct struct { Name string `xml:"Name"`