diff --git a/PLAN.md b/PLAN.md new file mode 100644 index 000000000..1c3343025 --- /dev/null +++ b/PLAN.md @@ -0,0 +1,145 @@ +# Plan: Implement CreateImage API + +## Context + +Users need the ability to create a new AMI from a running (or stopped) instance so they can launch new instances from snapshots of existing ones. This is the standard AWS `CreateImage` flow: snapshot the instance's root volume, then register a new AMI pointing at that snapshot. All the building blocks exist (CreateSnapshot, AMIMetadata, OpenFromSnapshot, cloneAMIToVolume) — they just need to be wired together. + +## Flow + +``` +AWS CLI: create-image --instance-id i-xxx --name "my-image" + → Gateway (ec2.go) + → NATS "ec2.CreateImage" + → Daemon handler + → Finds instance → gets root volume ID + → Calls snapshotService.CreateSnapshot(volumeID) + → Creates new AMI viperblock config with AMIMetadata.SnapshotID + → Stores as ami-xxx/config.json in S3 + → Returns ami-xxx +``` + +## Changes + +### 1. Gateway: Add CreateImage action entry +**File**: `hive/gateway/ec2.go` +- Add `"CreateImage"` to the `ec2Actions` map, following the DescribeImages pattern + +### 2. Gateway: CreateImage handler +**File**: `hive/gateway/ec2/image/CreateImage.go` (new) +- `ValidateCreateImageInput()` — validate InstanceId is present and starts with `i-` +- `CreateImage()` — create NATSImageService, call `svc.CreateImage(input)` +- Follow exact pattern of `gateway/ec2/snapshot/CreateSnapshot.go` + +### 3. NATS service: Wire CreateImage to NATS +**File**: `hive/handlers/ec2/image/service_nats.go` +- Replace the `CreateImage` stub with a real NATS request to `"ec2.CreateImage"` topic +- Use `utils.NATSRequest[ec2.CreateImageOutput]` with 120s timeout (snapshot can be slow) + +### 4. Daemon: Add NATS subscription and handler +**File**: `hive/daemon/daemon.go` +- Add `{"ec2.CreateImage", d.handleEC2CreateImage, "hive-workers"}` to subscriptions +- Update `NewImageServiceImpl(d.config)` → `NewImageServiceImpl(d.config, d.natsConn)` (~line 515) + +**File**: `hive/daemon/daemon_handlers.go` +- Add `handleEC2CreateImage` — this is a **stateful handler** (NOT handleNATSRequest) because it needs instance state to find the root volume +- Steps: + 1. Unmarshal `ec2.CreateImageInput` + 2. Lock `d.Instances.Mu`, find instance in `d.Instances.VMS[*input.InstanceId]`, unlock + 3. Validate instance exists and is in running or stopped state + 4. Find root volume ID from `instance.Instance.BlockDeviceMappings` (first entry with `Ebs.VolumeId`) + 5. Get instance status (running vs stopped) — pass to service so it knows which snapshot path to use + 6. Get source AMI ID from `instance.Instance.ImageId` (for architecture, platform, etc.) + 7. Call `d.imageService.CreateImageFromInstance(params)` passing the extracted data + 8. Respond with JSON + +### 5. Image service: Update interface and add CreateImageParams +**File**: `hive/handlers/ec2/image/service.go` +- Keep `CreateImage(input *ec2.CreateImageInput)` signature unchanged on the interface (gateway NATS side doesn't use extra params) + +**File**: `hive/handlers/ec2/image/service_impl.go` +- Add a `CreateImageParams` struct for the daemon-side call: + ```go + type CreateImageParams struct { + Input *ec2.CreateImageInput + RootVolumeID string + SourceImageID string + IsRunning bool // true = use ebs.snapshot NATS, false = snapshot from S3 state + } + ``` +- The daemon handler calls `d.imageService.CreateImageFromInstance(params)` directly (not through the generic interface) + +### 6. Image service: Implement CreateImageFromInstance +**File**: `hive/handlers/ec2/image/service_impl.go` + +Add `natsConn *nats.Conn` to `ImageServiceImpl` (for running-instance NATS snapshot calls and offline viperblock instantiation for stopped instances). Update constructor. + +Implement `CreateImageFromInstance(params CreateImageParams)`: +1. Generate AMI ID: `utils.GenerateResourceID("ami")` +2. Generate snapshot ID: `utils.GenerateResourceID("snap")` +3. **Snapshot the root volume** — two paths: + - **Running instance**: NATS `ebs.snapshot.{volumeID}` — reuse the exact pattern from `snapshot/service_impl.go:191-218`: + - Marshal `config.EBSSnapshotRequest{Volume: rootVolumeID, SnapshotID: snapshotID}` + - Send NATS request to `ebs.snapshot.{rootVolumeID}` with 30s timeout + - Validate response + - **Stopped instance**: Volume is not mounted in viperblockd. Data is fully persisted in S3. Create snapshot offline: + - Instantiate viperblock via `newViperblock(rootVolumeID, ...)` (reuse pattern from `instance/service_impl.go:238-258`) + - `backend.Init()` + `LoadStateRequest("")` to load config + block map from S3 + - Call `vb.CreateSnapshot(snapshotID)` directly — the flush/WAL steps are no-ops on a freshly loaded instance, it just serializes the block map into the snapshot checkpoint + - `vb.RemoveLocalFiles()` to clean up temp files +4. **Store snapshot metadata** in `{snapshotID}/metadata.json` — same pattern as `snapshot/service_impl.go:220-252` +5. **Read source AMI config** from S3 `{sourceImageID}/config.json` to get architecture, platform, etc. +6. **Read root volume config** from S3 `{rootVolumeID}/config.json` to get volume size +7. **Build AMI VolumeConfig** with populated `AMIMetadata`: + - `ImageID`: new ami-xxx + - `Name`: from input + - `Description`: from input + - `SnapshotID`: the snap-xxx just created + - `Architecture`, `PlatformDetails`, `Virtualization`, etc.: from source AMI (or defaults) + - `VolumeSizeGiB`: from root volume config + - `CreationDate`: now + - `RootDeviceType`: "ebs" + - `ImageOwnerAlias`: "self" +8. **Store AMI config** to S3 as `{amiID}/config.json` — create a `VBState` with the VolumeConfig. This is what `DescribeImages` reads. +9. Return `&ec2.CreateImageOutput{ImageId: &amiID}` + +### 7. Tests +- Add unit test for `CreateImage` in `hive/handlers/ec2/image/` with mock object store +- Add gateway validation test in `hive/gateway/ec2/image/` + +## Key Files + +| File | Change | +|------|--------| +| `hive/gateway/ec2.go` | Add CreateImage action | +| `hive/gateway/ec2/image/CreateImage.go` | New gateway handler | +| `hive/handlers/ec2/image/service.go` | Update interface | +| `hive/handlers/ec2/image/service_nats.go` | Wire NATS request | +| `hive/handlers/ec2/image/service_impl.go` | Core implementation | +| `hive/daemon/daemon.go` | Add subscription, update constructor | +| `hive/daemon/daemon_handlers.go` | Add stateful handler | + +## Existing Code to Reuse + +- `snapshot/service_impl.go:191-218` — EBS snapshot NATS pattern (`ebs.snapshot.{volumeID}`) +- `snapshot/service_impl.go:107-123` — `putSnapshotConfig` S3 storage pattern +- `image/service_impl.go:96-126` — Reading viperblock VBState from S3 (`config.json`) +- `instance/service_impl.go:238-258` — `newViperblock()` for offline viperblock instantiation +- `utils.GenerateResourceID()` — ID generation +- `utils.NATSRequest[T]()` — Generic NATS request helper +- `viperblock.VBState`, `viperblock.VolumeConfig`, `viperblock.AMIMetadata` — Existing types + +## Verification + +1. `make build` — confirm compilation +2. `make test` — all existing tests pass +3. Manual test: + ```bash + # Launch instance from existing AMI + aws ec2 run-instances --image-id ami-xxx --instance-type t2.micro --key-name mykey + # Create image from running instance + aws ec2 create-image --instance-id i-xxx --name "test-snapshot-image" + # Verify it appears + aws ec2 describe-images --owners self + # Launch new instance from the snapshot-based AMI + aws ec2 run-instances --image-id ami-yyy --instance-type t2.micro --key-name mykey + ``` diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index a1724cd7a..7e8521399 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -448,6 +448,7 @@ func (d *Daemon) subscribeAll() error { {"ec2.DescribeKeyPairs", d.handleEC2DescribeKeyPairs, "hive-workers"}, {"ec2.ImportKeyPair", d.handleEC2ImportKeyPair, "hive-workers"}, {"ec2.DescribeImages", d.handleEC2DescribeImages, "hive-workers"}, + {"ec2.CreateImage", d.handleEC2CreateImage, ""}, {"ec2.CreateVolume", d.handleEC2CreateVolume, "hive-workers"}, {"ec2.DescribeVolumes", d.handleEC2DescribeVolumes, "hive-workers"}, {"ec2.ModifyVolume", d.handleEC2ModifyVolume, "hive-workers"}, @@ -512,7 +513,7 @@ func (d *Daemon) Start() error { store := objectstore.NewS3ObjectStoreFromConfig(d.config.Predastore.Host, d.config.Predastore.Region, d.config.AccessKey, d.config.SecretKey) d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(d.config, d.resourceMgr.instanceTypes, d.natsConn, &d.Instances, store) d.keyService = handlers_ec2_key.NewKeyServiceImpl(d.config) - d.imageService = handlers_ec2_image.NewImageServiceImpl(d.config) + d.imageService = handlers_ec2_image.NewImageServiceImpl(d.config, d.natsConn) d.volumeService = handlers_ec2_volume.NewVolumeServiceImpl(d.config, d.natsConn) d.snapshotService = handlers_ec2_snapshot.NewSnapshotServiceImpl(d.config, d.natsConn) d.tagsService = handlers_ec2_tags.NewTagsServiceImpl(d.config) @@ -633,6 +634,37 @@ func (d *Daemon) restoreInstances() { } } + // Check if QEMU process is still alive from before the restart + if d.isInstanceProcessRunning(instance) { + slog.Info("Instance QEMU process still alive, reconnecting", "instance", instance.ID) + if err := d.reconnectInstance(instance); err != nil { + slog.Error("Failed to reconnect to running instance", "instanceId", instance.ID, "err", err) + } + continue + } + + // QEMU is not running - handle transitional states from interrupted operations + switch instance.Status { + case vm.StateStopping: + instance.Status = vm.StateStopped + slog.Info("Instance was stopping, QEMU exited, marking stopped", "instance", instance.ID) + if err := d.WriteState(); err != nil { + slog.Error("Failed to write state", "instance", instance.ID, "error", err) + } + continue + case vm.StateShuttingDown: + instance.Status = vm.StateTerminated + slog.Info("Instance was shutting-down, QEMU exited, marking terminated", "instance", instance.ID) + if err := d.WriteState(); err != nil { + slog.Error("Failed to write state", "instance", instance.ID, "error", err) + } + continue + case vm.StateRunning: + // Was running but QEMU died - reset to pending so LaunchInstance can transition to running + instance.Status = vm.StatePending + slog.Info("Instance was running but QEMU exited, relaunching", "instance", instance.ID) + } + slog.Info("Launching instance", "instance", instance.ID) if err := d.LaunchInstance(instance); err != nil { slog.Error("Failed to launch instance", "instanceId", instance.ID, "err", err) @@ -640,6 +672,46 @@ func (d *Daemon) restoreInstances() { } } +// isInstanceProcessRunning checks if the QEMU process for an instance is still alive. +func (d *Daemon) isInstanceProcessRunning(instance *vm.VM) bool { + pid, err := utils.ReadPidFile(instance.ID) + if err != nil || pid <= 0 { + return false + } + process, err := os.FindProcess(pid) + if err != nil { + return false + } + return process.Signal(syscall.Signal(0)) == nil +} + +// reconnectInstance re-establishes QMP and NATS connections to a running QEMU instance +// after a daemon restart. This bypasses the state machine since recovery is not a +// normal state transition. +func (d *Daemon) reconnectInstance(instance *vm.VM) error { + if err := d.CreateQMPClient(instance); err != nil { + return fmt.Errorf("failed to reconnect QMP: %w", err) + } + + d.mu.Lock() + sub, err := d.natsConn.QueueSubscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), "hive-events", d.handleEC2Events) + if err != nil { + d.mu.Unlock() + return fmt.Errorf("failed to subscribe to NATS: %w", err) + } + d.natsSubscriptions[instance.ID] = sub + d.mu.Unlock() + + instance.Status = vm.StateRunning + + if err := d.WriteState(); err != nil { + slog.Error("Failed to persist reconnected instance state", "instanceId", instance.ID, "err", err) + } + + slog.Info("Successfully reconnected to running instance", "instance", instance.ID) + return nil +} + // awaitShutdown blocks until the daemon's shutdown wait group completes. func (d *Daemon) awaitShutdown() { done := make(chan struct{}) diff --git a/hive/daemon/daemon_handlers.go b/hive/daemon/daemon_handlers.go index 78bb0d333..777186321 100644 --- a/hive/daemon/daemon_handlers.go +++ b/hive/daemon/daemon_handlers.go @@ -11,6 +11,7 @@ import ( "github.com/mulgadc/hive/hive/awserrors" "github.com/mulgadc/hive/hive/config" gateway_ec2_instance "github.com/mulgadc/hive/hive/gateway/ec2/instance" + handlers_ec2_image "github.com/mulgadc/hive/hive/handlers/ec2/image" "github.com/mulgadc/hive/hive/qmp" "github.com/mulgadc/hive/hive/utils" "github.com/mulgadc/hive/hive/vm" @@ -452,6 +453,17 @@ func (d *Daemon) handleDetachVolume(msg *nats.Msg, command qmp.Command, instance func (d *Daemon) handleStartInstance(msg *nats.Msg, command qmp.Command, instance *vm.VM, respondWithError func(string)) { slog.Info("Starting instance", "id", command.ID) + // Validate instance is in stopped state + d.Instances.Mu.Lock() + status := instance.Status + d.Instances.Mu.Unlock() + + if status != vm.StateStopped { + slog.Error("StartInstance: instance not in stopped state", "instanceId", command.ID, "status", status) + respondWithError(awserrors.ErrorIncorrectInstanceState) + return + } + // Allocate resources instanceType, ok := d.resourceMgr.instanceTypes[instance.InstanceType] if ok { @@ -814,6 +826,99 @@ func (d *Daemon) handleEC2DescribeImages(msg *nats.Msg) { handleNATSRequest(msg, d.imageService.DescribeImages) } +// handleEC2CreateImage is a stateful handler that extracts instance context +// (root volume ID, source AMI, running state) before delegating to the image service. +func (d *Daemon) handleEC2CreateImage(msg *nats.Msg) { + slog.Debug("Received message", "subject", msg.Subject) + + respondWithError := func(errCode string) { + if err := msg.Respond(utils.GenerateErrorPayload(errCode)); err != nil { + slog.Error("Failed to respond to NATS request", "err", err) + } + } + + input := &ec2.CreateImageInput{} + if errResp := utils.UnmarshalJsonPayload(input, msg.Data); errResp != nil { + if err := msg.Respond(errResp); err != nil { + slog.Error("Failed to respond to NATS request", "err", err) + } + return + } + + if input.InstanceId == nil || *input.InstanceId == "" { + respondWithError(awserrors.ErrorMissingParameter) + return + } + + instanceID := *input.InstanceId + + // Extract all instance context in a single critical section + d.Instances.Mu.Lock() + instance, ok := d.Instances.VMS[instanceID] + var status vm.InstanceState + var rootVolumeID, sourceImageID string + if ok { + status = instance.Status + if instance.Instance != nil { + for _, bdm := range instance.Instance.BlockDeviceMappings { + if bdm.Ebs != nil && bdm.Ebs.VolumeId != nil { + rootVolumeID = *bdm.Ebs.VolumeId + break + } + } + if instance.Instance.ImageId != nil { + sourceImageID = *instance.Instance.ImageId + } + } + } + d.Instances.Mu.Unlock() + + if !ok { + slog.Error("CreateImage: instance not found", "instanceId", instanceID) + respondWithError(awserrors.ErrorInvalidInstanceIDNotFound) + return + } + + if status != vm.StateRunning && status != vm.StateStopped { + slog.Error("CreateImage: instance not in valid state", "instanceId", instanceID, "status", status) + respondWithError(awserrors.ErrorIncorrectInstanceState) + return + } + + if rootVolumeID == "" { + slog.Error("CreateImage: no root volume found", "instanceId", instanceID) + respondWithError(awserrors.ErrorServerInternal) + return + } + + params := handlers_ec2_image.CreateImageParams{ + Input: input, + RootVolumeID: rootVolumeID, + SourceImageID: sourceImageID, + IsRunning: status == vm.StateRunning, + } + + output, err := d.imageService.CreateImageFromInstance(params) + if err != nil { + slog.Error("CreateImage: service failed", "instanceId", instanceID, "err", err) + respondWithError(err.Error()) + return + } + + jsonResponse, err := json.Marshal(output) + if err != nil { + slog.Error("CreateImage: failed to marshal response", "err", err) + respondWithError(awserrors.ErrorServerInternal) + return + } + + if err := msg.Respond(jsonResponse); err != nil { + slog.Error("Failed to respond to NATS request", "err", err) + } + + slog.Info("CreateImage completed", "instanceId", instanceID, "imageId", *output.ImageId) +} + func (d *Daemon) handleEC2CreateVolume(msg *nats.Msg) { handleNATSRequest(msg, d.volumeService.CreateVolume) } diff --git a/hive/daemon/daemon_handlers_test.go b/hive/daemon/daemon_handlers_test.go index 8c32c6f1e..bd2fd2df5 100644 --- a/hive/daemon/daemon_handlers_test.go +++ b/hive/daemon/daemon_handlers_test.go @@ -642,6 +642,148 @@ func TestHandleEC2DisableSerialConsoleAccess(t *testing.T) { assert.False(t, *output.SerialConsoleAccessEnabled) } +// --- handleEC2CreateImage tests --- + +func TestHandleEC2CreateImage_InstanceNotFound(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.Subscribe("ec2.CreateImage", daemon.handleEC2CreateImage) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.CreateImageInput{ + InstanceId: aws.String("i-nonexistent"), + Name: aws.String("my-image"), + } + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.CreateImage", reqData, 5*time.Second) + require.NoError(t, err) + + var errResp map[string]any + err = json.Unmarshal(reply.Data, &errResp) + require.NoError(t, err) + assert.Contains(t, errResp, "Code") +} + +func TestHandleEC2CreateImage_MissingInstanceId(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.Subscribe("ec2.CreateImage", daemon.handleEC2CreateImage) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.CreateImageInput{ + Name: aws.String("my-image"), + } + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.CreateImage", reqData, 5*time.Second) + require.NoError(t, err) + + var errResp map[string]any + err = json.Unmarshal(reply.Data, &errResp) + require.NoError(t, err) + assert.Contains(t, errResp, "Code") +} + +func TestHandleEC2CreateImage_InvalidState(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + // Add an instance in "pending" state (not running or stopped) + daemon.Instances.Mu.Lock() + daemon.Instances.VMS["i-pending123"] = &vm.VM{ + ID: "i-pending123", + Status: vm.StatePending, + Instance: &ec2.Instance{ + InstanceId: aws.String("i-pending123"), + ImageId: aws.String("ami-source"), + BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{ + { + DeviceName: aws.String("/dev/sda1"), + Ebs: &ec2.EbsInstanceBlockDevice{VolumeId: aws.String("vol-root123")}, + }, + }, + }, + } + daemon.Instances.Mu.Unlock() + + sub, err := daemon.natsConn.Subscribe("ec2.CreateImage", daemon.handleEC2CreateImage) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.CreateImageInput{ + InstanceId: aws.String("i-pending123"), + Name: aws.String("my-image"), + } + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.CreateImage", reqData, 5*time.Second) + require.NoError(t, err) + + var errResp map[string]any + err = json.Unmarshal(reply.Data, &errResp) + require.NoError(t, err) + assert.Contains(t, errResp, "Code") +} + +func TestHandleEC2CreateImage_NoRootVolume(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + // Add instance with no block device mappings + daemon.Instances.Mu.Lock() + daemon.Instances.VMS["i-novol123"] = &vm.VM{ + ID: "i-novol123", + Status: vm.StateRunning, + Instance: &ec2.Instance{ + InstanceId: aws.String("i-novol123"), + ImageId: aws.String("ami-source"), + BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{}, + }, + } + daemon.Instances.Mu.Unlock() + + sub, err := daemon.natsConn.Subscribe("ec2.CreateImage", daemon.handleEC2CreateImage) + require.NoError(t, err) + defer sub.Unsubscribe() + + input := &ec2.CreateImageInput{ + InstanceId: aws.String("i-novol123"), + Name: aws.String("my-image"), + } + reqData, _ := json.Marshal(input) + reply, err := daemon.natsConn.Request("ec2.CreateImage", reqData, 5*time.Second) + require.NoError(t, err) + + var errResp map[string]any + err = json.Unmarshal(reply.Data, &errResp) + require.NoError(t, err) + assert.Contains(t, errResp, "Code") +} + +func TestHandleEC2CreateImage_MalformedJSON(t *testing.T) { + natsURL := sharedNATSURL + + daemon := createFullTestDaemon(t, natsURL) + + sub, err := daemon.natsConn.Subscribe("ec2.CreateImage", daemon.handleEC2CreateImage) + require.NoError(t, err) + defer sub.Unsubscribe() + + reply, err := daemon.natsConn.Request("ec2.CreateImage", []byte(`{bad json}`), 5*time.Second) + require.NoError(t, err) + + var errResp map[string]any + err = json.Unmarshal(reply.Data, &errResp) + require.NoError(t, err) + assert.Contains(t, errResp, "Code") +} + // --- SetConfigPath test --- func TestSetConfigPath(t *testing.T) { diff --git a/hive/gateway/ec2.go b/hive/gateway/ec2.go index 4cf48744f..988cb5e31 100644 --- a/hive/gateway/ec2.go +++ b/hive/gateway/ec2.go @@ -87,6 +87,9 @@ var ec2Actions = map[string]EC2Handler{ "DescribeImages": ec2Handler(func(input *ec2.DescribeImagesInput, gw *GatewayConfig) (any, error) { return gateway_ec2_image.DescribeImages(input, gw.NATSConn) }), + "CreateImage": ec2Handler(func(input *ec2.CreateImageInput, gw *GatewayConfig) (any, error) { + return gateway_ec2_image.CreateImage(input, gw.NATSConn) + }), "DescribeRegions": ec2Handler(func(input *ec2.DescribeRegionsInput, gw *GatewayConfig) (any, error) { return gateway_ec2_zone.DescribeRegions(input, gw.Region) }), diff --git a/hive/gateway/ec2/image/CreateImage.go b/hive/gateway/ec2/image/CreateImage.go new file mode 100644 index 000000000..a28cc97ec --- /dev/null +++ b/hive/gateway/ec2/image/CreateImage.go @@ -0,0 +1,52 @@ +package gateway_ec2_image + +import ( + "errors" + "strings" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/hive/hive/awserrors" + handlers_ec2_image "github.com/mulgadc/hive/hive/handlers/ec2/image" + "github.com/nats-io/nats.go" +) + +// ValidateCreateImageInput validates the input parameters for CreateImage +func ValidateCreateImageInput(input *ec2.CreateImageInput) error { + if input == nil { + return errors.New(awserrors.ErrorInvalidParameterValue) + } + + if input.InstanceId == nil || *input.InstanceId == "" { + return errors.New(awserrors.ErrorMissingParameter) + } + + if !strings.HasPrefix(*input.InstanceId, "i-") { + return errors.New(awserrors.ErrorInvalidInstanceIDMalformed) + } + + if input.Name == nil || *input.Name == "" { + return errors.New(awserrors.ErrorMissingParameter) + } + + return nil +} + +// CreateImage handles the EC2 CreateImage API call +func CreateImage(input *ec2.CreateImageInput, natsConn *nats.Conn) (ec2.CreateImageOutput, error) { + var output ec2.CreateImageOutput + + if err := ValidateCreateImageInput(input); err != nil { + return output, err + } + + svc := handlers_ec2_image.NewNATSImageService(natsConn) + result, err := svc.CreateImage(input) + if err != nil { + return output, err + } + if result == nil { + return output, errors.New(awserrors.ErrorServerInternal) + } + + return *result, nil +} diff --git a/hive/gateway/ec2/image/CreateImage_test.go b/hive/gateway/ec2/image/CreateImage_test.go new file mode 100644 index 000000000..e2d419e2e --- /dev/null +++ b/hive/gateway/ec2/image/CreateImage_test.go @@ -0,0 +1,97 @@ +package gateway_ec2_image + +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 TestValidateCreateImageInput(t *testing.T) { + tests := []struct { + name string + input *ec2.CreateImageInput + wantErr bool + errMsg string + }{ + { + name: "NilInput", + input: nil, + wantErr: true, + errMsg: awserrors.ErrorInvalidParameterValue, + }, + { + name: "EmptyInput", + input: &ec2.CreateImageInput{}, + wantErr: true, + errMsg: awserrors.ErrorMissingParameter, + }, + { + name: "MissingName", + input: &ec2.CreateImageInput{ + InstanceId: aws.String("i-1234567890abcdef0"), + }, + wantErr: true, + errMsg: awserrors.ErrorMissingParameter, + }, + { + name: "InvalidInstanceIdFormat", + input: &ec2.CreateImageInput{ + InstanceId: aws.String("invalid-id"), + Name: aws.String("my-image"), + }, + wantErr: true, + errMsg: awserrors.ErrorInvalidInstanceIDMalformed, + }, + { + name: "EmptyInstanceId", + input: &ec2.CreateImageInput{ + InstanceId: aws.String(""), + Name: aws.String("my-image"), + }, + wantErr: true, + errMsg: awserrors.ErrorMissingParameter, + }, + { + name: "EmptyName", + input: &ec2.CreateImageInput{ + InstanceId: aws.String("i-1234567890abcdef0"), + Name: aws.String(""), + }, + wantErr: true, + errMsg: awserrors.ErrorMissingParameter, + }, + { + name: "ValidInput", + input: &ec2.CreateImageInput{ + InstanceId: aws.String("i-1234567890abcdef0"), + Name: aws.String("my-image"), + }, + wantErr: false, + }, + { + name: "ValidInputWithDescription", + input: &ec2.CreateImageInput{ + InstanceId: aws.String("i-1234567890abcdef0"), + Name: aws.String("my-image"), + Description: aws.String("A snapshot image"), + }, + wantErr: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := ValidateCreateImageInput(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/image/service_impl.go b/hive/handlers/ec2/image/service_impl.go index 82f528dab..2d924f137 100644 --- a/hive/handlers/ec2/image/service_impl.go +++ b/hive/handlers/ec2/image/service_impl.go @@ -1,35 +1,51 @@ package handlers_ec2_image import ( + "bytes" "encoding/json" "errors" + "fmt" "io" "log/slog" "strings" + "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" "github.com/aws/aws-sdk-go/service/s3" "github.com/mulgadc/hive/hive/awserrors" "github.com/mulgadc/hive/hive/config" + handlers_ec2_snapshot "github.com/mulgadc/hive/hive/handlers/ec2/snapshot" "github.com/mulgadc/hive/hive/objectstore" "github.com/mulgadc/hive/hive/utils" "github.com/mulgadc/viperblock/viperblock" + vbs3 "github.com/mulgadc/viperblock/viperblock/backends/s3" + "github.com/nats-io/nats.go" ) // Ensure ImageServiceImpl implements ImageService var _ ImageService = (*ImageServiceImpl)(nil) +// CreateImageParams holds parameters for creating an AMI from an instance. +// Used by the daemon handler which extracts instance state before calling the service. +type CreateImageParams struct { + Input *ec2.CreateImageInput + RootVolumeID string + SourceImageID string + IsRunning bool // true = use ebs.snapshot NATS, false = snapshot from S3 state +} + // ImageServiceImpl handles AMI image operations with S3 storage type ImageServiceImpl struct { config *config.Config store objectstore.ObjectStore + natsConn *nats.Conn bucketName string accountID string // AWS account ID for S3 key storage path } // NewImageServiceImpl creates a new daemon-side image service -func NewImageServiceImpl(cfg *config.Config) *ImageServiceImpl { +func NewImageServiceImpl(cfg *config.Config, natsConn *nats.Conn) *ImageServiceImpl { store := objectstore.NewS3ObjectStoreFromConfig( cfg.Predastore.Host, cfg.Predastore.Region, @@ -40,6 +56,7 @@ func NewImageServiceImpl(cfg *config.Config) *ImageServiceImpl { return &ImageServiceImpl{ config: cfg, store: store, + natsConn: natsConn, bucketName: cfg.Predastore.Bucket, accountID: "123456789", // TODO: Implement proper account ID management } @@ -215,9 +232,275 @@ func (s *ImageServiceImpl) DescribeImages(input *ec2.DescribeImagesInput) (*ec2. }, nil } -// Stub implementations for other ImageService methods +// CreateImage is the generic interface method — on the daemon side, the handler +// calls CreateImageFromInstance directly with the extra instance context. func (s *ImageServiceImpl) CreateImage(input *ec2.CreateImageInput) (*ec2.CreateImageOutput, error) { - return nil, errors.New("CreateImage not yet implemented") + return nil, errors.New("CreateImage requires instance context; use CreateImageFromInstance on daemon side") +} + +// CreateImageFromInstance creates an AMI from an instance by snapshotting the root +// volume and storing a new AMI config in S3. +func (s *ImageServiceImpl) CreateImageFromInstance(params CreateImageParams) (*ec2.CreateImageOutput, error) { + input := params.Input + if input == nil || input.InstanceId == nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + + amiID := utils.GenerateResourceID("ami") + snapshotID := utils.GenerateResourceID("snap") + + slog.Info("CreateImageFromInstance", "instanceId", *input.InstanceId, + "rootVolumeId", params.RootVolumeID, "amiId", amiID, "snapshotId", snapshotID, + "isRunning", params.IsRunning) + + // Step 1: Snapshot the root volume + if params.IsRunning { + if err := s.snapshotRunningVolume(params.RootVolumeID, snapshotID); err != nil { + return nil, err + } + } else { + if err := s.snapshotStoppedVolume(params.RootVolumeID, snapshotID); err != nil { + return nil, err + } + } + + // Step 2: Read source volume config for size + volumeConfig, err := s.getVolumeConfig(params.RootVolumeID) + if err != nil { + slog.Error("CreateImageFromInstance: failed to read volume config", "volumeId", params.RootVolumeID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + volumeSizeGiB := volumeConfig.VolumeMetadata.SizeGiB + + // Step 3: Read source AMI config for architecture, platform, etc. + var sourceAMI viperblock.AMIMetadata + if params.SourceImageID != "" { + srcCfg, err := s.getAMIConfig(params.SourceImageID) + if err != nil { + slog.Error("CreateImageFromInstance: failed to read source AMI config", "sourceImageId", params.SourceImageID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + sourceAMI = srcCfg + } else { + sourceAMI = viperblock.AMIMetadata{ + Architecture: "x86_64", + PlatformDetails: "Linux/UNIX", + Virtualization: "hvm", + } + } + + // Step 4: Store snapshot metadata + if err := s.putSnapshotMetadata(snapshotID, params.RootVolumeID, volumeSizeGiB); err != nil { + slog.Error("CreateImageFromInstance: failed to write snapshot metadata", "snapshotId", snapshotID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + + // Step 5: Build and store AMI config + name := aws.StringValue(input.Name) + description := aws.StringValue(input.Description) + + amiConfig := viperblock.VBState{ + VolumeConfig: viperblock.VolumeConfig{ + AMIMetadata: viperblock.AMIMetadata{ + ImageID: amiID, + Name: name, + Description: description, + SnapshotID: snapshotID, + Architecture: sourceAMI.Architecture, + PlatformDetails: sourceAMI.PlatformDetails, + Virtualization: sourceAMI.Virtualization, + VolumeSizeGiB: volumeSizeGiB, + CreationDate: time.Now(), + RootDeviceType: "ebs", + ImageOwnerAlias: "self", + }, + }, + } + + configData, err := json.Marshal(amiConfig) + if err != nil { + slog.Error("CreateImageFromInstance: failed to marshal AMI config", "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + + configKey := fmt.Sprintf("%s/config.json", amiID) + _, err = s.store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String(s.bucketName), + Key: aws.String(configKey), + Body: bytes.NewReader(configData), + ContentType: aws.String("application/json"), + }) + if err != nil { + slog.Error("CreateImageFromInstance: failed to store AMI config", "amiId", amiID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + + slog.Info("CreateImageFromInstance completed", "amiId", amiID, "snapshotId", snapshotID) + + return &ec2.CreateImageOutput{ + ImageId: aws.String(amiID), + }, nil +} + +// snapshotRunningVolume triggers a snapshot via NATS on a running viperblockd instance +func (s *ImageServiceImpl) snapshotRunningVolume(volumeID, snapshotID string) error { + if s.natsConn == nil { + return errors.New(awserrors.ErrorServerInternal) + } + + snapReq := config.EBSSnapshotRequest{Volume: volumeID, SnapshotID: snapshotID} + snapData, err := json.Marshal(snapReq) + if err != nil { + slog.Error("snapshotRunningVolume: failed to marshal snapshot request", "volumeId", volumeID, "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + + msg, err := s.natsConn.Request(fmt.Sprintf("ebs.snapshot.%s", volumeID), snapData, 30*time.Second) + if err != nil { + slog.Error("snapshotRunningVolume: NATS request failed", "volumeId", volumeID, "snapshotId", snapshotID, "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + + var snapResp config.EBSSnapshotResponse + if err := json.Unmarshal(msg.Data, &snapResp); err != nil { + slog.Error("snapshotRunningVolume: failed to unmarshal response", "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + if !snapResp.Success || snapResp.Error != "" { + slog.Error("snapshotRunningVolume: snapshot failed", "volumeId", volumeID, "err", snapResp.Error) + return errors.New(awserrors.ErrorServerInternal) + } + + slog.Info("snapshotRunningVolume: snapshot created", "volumeId", volumeID, "snapshotId", snapshotID) + return nil +} + +// snapshotStoppedVolume creates a snapshot offline by loading viperblock state from S3 +func (s *ImageServiceImpl) snapshotStoppedVolume(volumeID, snapshotID string) error { + if s.config == nil { + return errors.New(awserrors.ErrorServerInternal) + } + + // Read volume config to get the size (required by viperblock.New) + volConfig, err := s.getVolumeConfig(volumeID) + if err != nil { + slog.Error("snapshotStoppedVolume: failed to read volume config", "volumeId", volumeID, "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + volumeSize := volConfig.VolumeMetadata.SizeGiB * 1024 * 1024 * 1024 + + cfg := vbs3.S3Config{ + VolumeName: volumeID, + VolumeSize: volumeSize, + Bucket: s.config.Predastore.Bucket, + Region: s.config.Predastore.Region, + AccessKey: s.config.Predastore.AccessKey, + SecretKey: s.config.Predastore.SecretKey, + Host: s.config.Predastore.Host, + } + + vbconfig := viperblock.VB{ + VolumeName: volumeID, + VolumeSize: volumeSize, + BaseDir: s.config.WalDir, + Cache: viperblock.Cache{Config: viperblock.CacheConfig{Size: 0}}, + } + + vb, err := viperblock.New(&vbconfig, "s3", cfg) + if err != nil { + slog.Error("snapshotStoppedVolume: failed to create viperblock instance", "volumeId", volumeID, "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + + if err := vb.Backend.Init(); err != nil { + slog.Error("snapshotStoppedVolume: failed to init backend", "volumeId", volumeID, "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + + if _, err := vb.LoadStateRequest(""); err != nil { + slog.Error("snapshotStoppedVolume: failed to load state", "volumeId", volumeID, "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + + if _, err := vb.CreateSnapshot(snapshotID); err != nil { + slog.Error("snapshotStoppedVolume: CreateSnapshot failed", "volumeId", volumeID, "snapshotId", snapshotID, "err", err) + if cleanupErr := vb.RemoveLocalFiles(); cleanupErr != nil { + slog.Error("snapshotStoppedVolume: failed to remove local files after snapshot failure", "volumeId", volumeID, "err", cleanupErr) + } + return errors.New(awserrors.ErrorServerInternal) + } + + if err := vb.RemoveLocalFiles(); err != nil { + slog.Error("snapshotStoppedVolume: failed to remove local files", "volumeId", volumeID, "err", err) + } + + slog.Info("snapshotStoppedVolume: snapshot created", "volumeId", volumeID, "snapshotId", snapshotID) + return nil +} + +// getVolumeConfig reads a volume's VBState config from S3 +func (s *ImageServiceImpl) getVolumeConfig(volumeID string) (*viperblock.VolumeConfig, error) { + configKey := fmt.Sprintf("%s/config.json", volumeID) + result, err := s.store.GetObject(&s3.GetObjectInput{ + Bucket: aws.String(s.bucketName), + Key: aws.String(configKey), + }) + if err != nil { + return nil, err + } + defer result.Body.Close() + + var vbState viperblock.VBState + if err := json.NewDecoder(result.Body).Decode(&vbState); err != nil { + return nil, err + } + return &vbState.VolumeConfig, nil +} + +// getAMIConfig reads an AMI's metadata from S3 +func (s *ImageServiceImpl) getAMIConfig(imageID string) (viperblock.AMIMetadata, error) { + configKey := fmt.Sprintf("%s/config.json", imageID) + result, err := s.store.GetObject(&s3.GetObjectInput{ + Bucket: aws.String(s.bucketName), + Key: aws.String(configKey), + }) + if err != nil { + return viperblock.AMIMetadata{}, err + } + defer result.Body.Close() + + var vbState viperblock.VBState + if err := json.NewDecoder(result.Body).Decode(&vbState); err != nil { + return viperblock.AMIMetadata{}, err + } + return vbState.VolumeConfig.AMIMetadata, nil +} + +// putSnapshotMetadata stores snapshot metadata in S3 using the canonical SnapshotConfig type +func (s *ImageServiceImpl) putSnapshotMetadata(snapshotID, volumeID string, volumeSizeGiB uint64) error { + cfg := handlers_ec2_snapshot.SnapshotConfig{ + SnapshotID: snapshotID, + VolumeID: volumeID, + VolumeSize: utils.SafeUint64ToInt64(volumeSizeGiB), + State: "completed", + Progress: "100%", + StartTime: time.Now(), + OwnerID: s.accountID, + } + + data, err := json.Marshal(cfg) + if err != nil { + return err + } + + key := fmt.Sprintf("%s/metadata.json", snapshotID) + _, err = s.store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String(s.bucketName), + Key: aws.String(key), + Body: bytes.NewReader(data), + ContentType: aws.String("application/json"), + }) + return err } func (s *ImageServiceImpl) CopyImage(input *ec2.CopyImageInput) (*ec2.CopyImageOutput, error) { diff --git a/hive/handlers/ec2/image/service_impl_test.go b/hive/handlers/ec2/image/service_impl_test.go new file mode 100644 index 000000000..9592929ee --- /dev/null +++ b/hive/handlers/ec2/image/service_impl_test.go @@ -0,0 +1,261 @@ +package handlers_ec2_image + +import ( + "encoding/json" + "strings" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + awss3 "github.com/aws/aws-sdk-go/service/s3" + "github.com/mulgadc/hive/hive/awserrors" + handlers_ec2_snapshot "github.com/mulgadc/hive/hive/handlers/ec2/snapshot" + "github.com/mulgadc/hive/hive/objectstore" + "github.com/mulgadc/viperblock/viperblock" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const testBucket = "test-bucket" +const testAccountID = "123456789" + +// setupTestImageService creates an image service with in-memory storage for testing +func setupTestImageService(t *testing.T) (*ImageServiceImpl, *objectstore.MemoryObjectStore) { + store := objectstore.NewMemoryObjectStore() + svc := NewImageServiceImplWithStore(store, testBucket, testAccountID) + return svc, store +} + +// createTestVolumeConfig creates a test volume config in the mock store +func createTestVolumeConfig(t *testing.T, store *objectstore.MemoryObjectStore, volumeID string, sizeGiB int) { + volumeState := viperblock.VBState{ + VolumeConfig: viperblock.VolumeConfig{ + VolumeMetadata: viperblock.VolumeMetadata{ + SizeGiB: uint64(sizeGiB), + }, + }, + } + data, err := json.Marshal(volumeState) + require.NoError(t, err) + + _, err = store.PutObject(&awss3.PutObjectInput{ + Bucket: aws.String(testBucket), + Key: aws.String(volumeID + "/config.json"), + Body: strings.NewReader(string(data)), + ContentType: aws.String("application/json"), + }) + require.NoError(t, err) +} + +// createTestAMIConfig creates a test AMI config in the mock store +func createTestAMIConfig(t *testing.T, store *objectstore.MemoryObjectStore, imageID string) { + amiState := viperblock.VBState{ + VolumeConfig: viperblock.VolumeConfig{ + AMIMetadata: viperblock.AMIMetadata{ + ImageID: imageID, + Name: "test-ami", + Architecture: "x86_64", + PlatformDetails: "Linux/UNIX", + Virtualization: "hvm", + RootDeviceType: "ebs", + VolumeSizeGiB: 8, + }, + }, + } + data, err := json.Marshal(amiState) + require.NoError(t, err) + + _, err = store.PutObject(&awss3.PutObjectInput{ + Bucket: aws.String(testBucket), + Key: aws.String(imageID + "/config.json"), + Body: strings.NewReader(string(data)), + ContentType: aws.String("application/json"), + }) + require.NoError(t, err) +} + +func TestCreateImageFromInstance_NilInput(t *testing.T) { + svc, _ := setupTestImageService(t) + + _, err := svc.CreateImageFromInstance(CreateImageParams{}) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error()) +} + +func TestCreateImageFromInstance_RunningInstance_NoNATS(t *testing.T) { + svc, store := setupTestImageService(t) + + // Create volume and AMI configs + createTestVolumeConfig(t, store, "vol-root123", 10) + createTestAMIConfig(t, store, "ami-source123") + + // Running instance without NATS should fail (natsConn is nil) + _, err := svc.CreateImageFromInstance(CreateImageParams{ + Input: &ec2.CreateImageInput{ + InstanceId: aws.String("i-test123"), + Name: aws.String("my-image"), + }, + RootVolumeID: "vol-root123", + SourceImageID: "ami-source123", + IsRunning: true, + }) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) +} + +func TestCreateImageFromInstance_StoppedInstance_NoConfig(t *testing.T) { + svc, _ := setupTestImageService(t) + + // Stopped instance without config will fail (no config to create viperblock) + _, err := svc.CreateImageFromInstance(CreateImageParams{ + Input: &ec2.CreateImageInput{ + InstanceId: aws.String("i-test123"), + Name: aws.String("my-image"), + }, + RootVolumeID: "vol-nonexistent", + SourceImageID: "ami-source123", + IsRunning: false, + }) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) +} + +func TestDescribeImages_AfterCreate(t *testing.T) { + svc, store := setupTestImageService(t) + + // Manually create an AMI config in the store (simulates what CreateImageFromInstance does) + amiID := "ami-testimage123" + createTestAMIConfig(t, store, amiID) + + // Describe images should find it + result, err := svc.DescribeImages(&ec2.DescribeImagesInput{ + ImageIds: []*string{aws.String(amiID)}, + }) + require.NoError(t, err) + require.NotNil(t, result) + require.Len(t, result.Images, 1) + assert.Equal(t, amiID, *result.Images[0].ImageId) + assert.Equal(t, "test-ami", *result.Images[0].Name) + assert.Equal(t, "x86_64", *result.Images[0].Architecture) +} + +func TestGetVolumeConfig(t *testing.T) { + svc, store := setupTestImageService(t) + + createTestVolumeConfig(t, store, "vol-abc123", 20) + + cfg, err := svc.getVolumeConfig("vol-abc123") + require.NoError(t, err) + require.NotNil(t, cfg) + assert.Equal(t, uint64(20), cfg.VolumeMetadata.SizeGiB) +} + +func TestGetVolumeConfig_NotFound(t *testing.T) { + svc, _ := setupTestImageService(t) + + _, err := svc.getVolumeConfig("vol-nonexistent") + require.Error(t, err) +} + +func TestGetAMIConfig(t *testing.T) { + svc, store := setupTestImageService(t) + + createTestAMIConfig(t, store, "ami-abc123") + + meta, err := svc.getAMIConfig("ami-abc123") + require.NoError(t, err) + assert.Equal(t, "ami-abc123", meta.ImageID) + assert.Equal(t, "test-ami", meta.Name) + assert.Equal(t, "x86_64", meta.Architecture) + assert.Equal(t, "Linux/UNIX", meta.PlatformDetails) + assert.Equal(t, "hvm", meta.Virtualization) +} + +func TestGetAMIConfig_NotFound(t *testing.T) { + svc, _ := setupTestImageService(t) + + _, err := svc.getAMIConfig("ami-nonexistent") + require.Error(t, err) +} + +func TestPutSnapshotMetadata(t *testing.T) { + svc, store := setupTestImageService(t) + + err := svc.putSnapshotMetadata("snap-abc123", "vol-xyz789", 10) + require.NoError(t, err) + + // Verify the metadata was written correctly + result, err := store.GetObject(&awss3.GetObjectInput{ + Bucket: aws.String(testBucket), + Key: aws.String("snap-abc123/metadata.json"), + }) + require.NoError(t, err) + defer result.Body.Close() + + var cfg handlers_ec2_snapshot.SnapshotConfig + err = json.NewDecoder(result.Body).Decode(&cfg) + require.NoError(t, err) + assert.Equal(t, "snap-abc123", cfg.SnapshotID) + assert.Equal(t, "vol-xyz789", cfg.VolumeID) + assert.Equal(t, int64(10), cfg.VolumeSize) + assert.Equal(t, "completed", cfg.State) + assert.Equal(t, "100%", cfg.Progress) + assert.Equal(t, testAccountID, cfg.OwnerID) +} + +func TestCreateImageFromInstance_SourceAMIReadFailure(t *testing.T) { + svc, store := setupTestImageService(t) + + // Create volume config but NOT the source AMI config + createTestVolumeConfig(t, store, "vol-root123", 10) + + // With non-empty SourceImageID, missing AMI config should be a hard error + _, err := svc.CreateImageFromInstance(CreateImageParams{ + Input: &ec2.CreateImageInput{ + InstanceId: aws.String("i-test123"), + Name: aws.String("my-image"), + }, + RootVolumeID: "vol-root123", + SourceImageID: "ami-nonexistent", + IsRunning: true, // will fail at snapshot step first (no NATS) + }) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) +} + +func TestDescribeImages_FilterByOwnerSelf(t *testing.T) { + svc, store := setupTestImageService(t) + + // Create AMI with "self" owner alias + amiID := "ami-selfowned" + amiState := viperblock.VBState{ + VolumeConfig: viperblock.VolumeConfig{ + AMIMetadata: viperblock.AMIMetadata{ + ImageID: amiID, + Name: "self-owned-ami", + ImageOwnerAlias: "self", + Architecture: "x86_64", + PlatformDetails: "Linux/UNIX", + RootDeviceType: "ebs", + VolumeSizeGiB: 8, + }, + }, + } + data, err := json.Marshal(amiState) + require.NoError(t, err) + _, err = store.PutObject(&awss3.PutObjectInput{ + Bucket: aws.String(testBucket), + Key: aws.String(amiID + "/config.json"), + Body: strings.NewReader(string(data)), + }) + require.NoError(t, err) + + // Filter by "self" + result, err := svc.DescribeImages(&ec2.DescribeImagesInput{ + Owners: []*string{aws.String("self")}, + }) + require.NoError(t, err) + require.NotNil(t, result) + assert.Len(t, result.Images, 1) + assert.Equal(t, amiID, *result.Images[0].ImageId) +} diff --git a/hive/handlers/ec2/image/service_nats.go b/hive/handlers/ec2/image/service_nats.go index 1d949b199..307bf7253 100644 --- a/hive/handlers/ec2/image/service_nats.go +++ b/hive/handlers/ec2/image/service_nats.go @@ -1,7 +1,7 @@ package handlers_ec2_image import ( - "fmt" + "errors" "time" "github.com/aws/aws-sdk-go/service/ec2" @@ -23,31 +23,30 @@ func (s *NATSImageService) DescribeImages(input *ec2.DescribeImagesInput) (*ec2. return utils.NATSRequest[ec2.DescribeImagesOutput](s.natsConn, "ec2.DescribeImages", input, 30*time.Second) } -// Stub implementations for unimplemented ImageService methods func (s *NATSImageService) CreateImage(input *ec2.CreateImageInput) (*ec2.CreateImageOutput, error) { - return nil, fmt.Errorf("CreateImage not yet implemented") + return utils.NATSRequest[ec2.CreateImageOutput](s.natsConn, "ec2.CreateImage", input, 120*time.Second) } func (s *NATSImageService) CopyImage(input *ec2.CopyImageInput) (*ec2.CopyImageOutput, error) { - return nil, fmt.Errorf("CopyImage not yet implemented") + return nil, errors.New("CopyImage not yet implemented") } func (s *NATSImageService) DescribeImageAttribute(input *ec2.DescribeImageAttributeInput) (*ec2.DescribeImageAttributeOutput, error) { - return nil, fmt.Errorf("DescribeImageAttribute not yet implemented") + return nil, errors.New("DescribeImageAttribute not yet implemented") } func (s *NATSImageService) RegisterImage(input *ec2.RegisterImageInput) (*ec2.RegisterImageOutput, error) { - return nil, fmt.Errorf("RegisterImage not yet implemented") + return nil, errors.New("RegisterImage not yet implemented") } func (s *NATSImageService) DeregisterImage(input *ec2.DeregisterImageInput) (*ec2.DeregisterImageOutput, error) { - return nil, fmt.Errorf("DeregisterImage not yet implemented") + return nil, errors.New("DeregisterImage not yet implemented") } func (s *NATSImageService) ModifyImageAttribute(input *ec2.ModifyImageAttributeInput) (*ec2.ModifyImageAttributeOutput, error) { - return nil, fmt.Errorf("ModifyImageAttribute not yet implemented") + return nil, errors.New("ModifyImageAttribute not yet implemented") } func (s *NATSImageService) ResetImageAttribute(input *ec2.ResetImageAttributeInput) (*ec2.ResetImageAttributeOutput, error) { - return nil, fmt.Errorf("ResetImageAttribute not yet implemented") + return nil, errors.New("ResetImageAttribute not yet implemented") } diff --git a/scripts/stop-dev.sh b/scripts/stop-dev.sh index 7ea7e861a..515e5faeb 100755 --- a/scripts/stop-dev.sh +++ b/scripts/stop-dev.sh @@ -65,6 +65,27 @@ stop_service "hive-ui" "$PID_DIR" # Stop Hive daemon/gateway (it will terminate running instances, unmount nbd devices) stop_service "hive" "$PID_DIR" +# Wait for all QEMU instances to exit before stopping infrastructure services. +# The daemon sends system_powerdown to VMs during shutdown, but may be killed +# before VMs fully exit. We must wait here so viperblock/predastore aren't +# pulled out from under running VMs. +if pgrep -x qemu-system-x86_64 > /dev/null 2>&1; then + echo "⏳ Waiting for QEMU instances to exit..." + timeout=120 + elapsed=0 + while pgrep -x qemu-system-x86_64 > /dev/null 2>&1; do + if [ $elapsed -ge $timeout ]; then + echo "⚠️ Timeout waiting for QEMU processes, force killing..." + pkill -9 -x qemu-system-x86_64 2>/dev/null || true + sleep 1 + break + fi + sleep 1 + ((elapsed++)) + done + echo "✅ All QEMU instances exited" +fi + # Stop AWSGW stop_service "awsgw" "$PID_DIR"