mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-21 11:46:37 +00:00
feat: attach-volume (#20)
This commit is contained in:
@@ -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<br>2. Modify volume type<br>3. Decrease size (error - not supported)<br>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<br>2. Boundary sizes (1 GiB min, 16384 GiB max)<br>3. Invalid AZ (error)<br>4. Verify volume in describe-volumes<br>5. Unsupported volume type (error - only gp3)<br>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<br>2. Delete attached volume (error: VolumeInUse)<br>3. Delete non-existent volume (error: InvalidVolume.NotFound)<br>4. Verify volume gone from describe-volumes<br>5. Malformed volume ID (error: InvalidVolumeID.Malformed)<br>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<br>2. Verify device visible in guest OS<br>3. Attach already-attached volume (error)<br>4. Attach to non-existent instance (error)<br>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<br>2. Auto-assign device name<br>3. Attach already-attached volume (VolumeInUse)<br>4. Attach to non-existent instance (InvalidInstanceID.NotFound)<br>5. Attach to stopped instance (IncorrectInstanceState)<br>6. Volume not found (InvalidVolume.NotFound)<br>7. All device slots full (AttachmentLimitExceeded)<br>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<br>2. Force detach<br>3. Detach already-detached volume (error)<br>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<br>2. Check volume during modification<br>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<br>2. Check completed modification<br>3. No modifications returns empty | **NOT STARTED** |
|
||||
|
||||
@@ -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.
|
||||
@@ -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
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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
|
||||
|
||||
+75
-4
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
+10
-3
@@ -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
|
||||
|
||||
@@ -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=<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/
|
||||
|
||||
@@ -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"`
|
||||
|
||||
Reference in New Issue
Block a user