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

feat: more awserrors (#46)

Browse Source
This commit is contained in:
Julian Sommer
2026-02-19 12:15:45 +11:00
committed by GitHub
parent 8573800914
commit a5ff1ca5a7
12 changed files with 821 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hive/daemon/daemon_handlers.go
+66 -4
View File
@@ -135,6 +135,16 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command qmp.Command, instance
return
}
// Check AZ compatibility — volume and instance must be in the same AZ
if volCfg.VolumeMetadata.AvailabilityZone != "" && d.config.AZ != "" &&
volCfg.VolumeMetadata.AvailabilityZone != d.config.AZ {
slog.Error("AttachVolume: volume and instance are in different AZs",
"volumeId", volumeID, "volumeAZ", volCfg.VolumeMetadata.AvailabilityZone,
"instanceAZ", d.config.AZ)
respondWithError(awserrors.ErrorInvalidVolumeZoneMismatch)
return
}
// Determine device name
if device == "" {
d.Instances.Mu.Lock()
@@ -545,6 +555,18 @@ func (d *Daemon) handleStopOrTerminateInstance(msg *nats.Msg, command qmp.Comman
slog.Info(action+" instance", "id", command.ID)
// Check state validity before attempting transition — return the correct
// AWS error code when the instance is already stopped/terminated/etc.
d.Instances.Mu.Lock()
currentState := instance.Status
d.Instances.Mu.Unlock()
if !vm.IsValidTransition(currentState, initialState) {
slog.Warn("Instance in incorrect state for "+strings.ToLower(action),
"instanceId", instance.ID, "currentState", string(currentState))
respondWithError(awserrors.ErrorIncorrectInstanceState)
return
}
// Transition to the initial transitional state
if err := d.TransitionState(instance, initialState); err != nil {
slog.Error("Failed to transition to "+string(initialState), "instanceId", instance.ID, "err", err)
@@ -709,6 +731,31 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
return
}
// Validate AMI exists before allocating resources
if runInstancesInput.ImageId != nil {
_, err := d.imageService.GetAMIConfig(*runInstancesInput.ImageId)
if err != nil {
slog.Error("handleEC2RunInstances AMI not found", "imageId", *runInstancesInput.ImageId, "err", err)
errResp = utils.GenerateErrorPayload(awserrors.ErrorInvalidAMIIDNotFound)
if err := msg.Respond(errResp); err != nil {
slog.Error("Failed to respond to NATS request", "err", err)
}
return
}
}
// Validate key pair exists (if specified)
if runInstancesInput.KeyName != nil && *runInstancesInput.KeyName != "" {
if err := d.keyService.ValidateKeyPairExists(*runInstancesInput.KeyName); err != nil {
slog.Error("handleEC2RunInstances key pair not found", "keyName", *runInstancesInput.KeyName, "err", err)
errResp = utils.GenerateErrorPayload(awserrors.ErrorInvalidKeyPairNotFound)
if err := msg.Respond(errResp); err != nil {
slog.Error("Failed to respond to NATS request", "err", err)
}
return
}
}
// Determine how many instances to launch based on MinCount/MaxCount
minCount := int(*runInstancesInput.MinCount)
maxCount := int(*runInstancesInput.MaxCount)
@@ -769,11 +816,13 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
// Create all instances
var instances []*vm.VM
var allEC2Instances []*ec2.Instance
var lastRunErr error
for i := 0; i < launchCount; i++ {
instance, ec2Instance, err := d.instanceService.RunInstance(runInstancesInput)
if err != nil {
slog.Error("handleEC2RunInstances service.RunInstance failed", "index", i, "err", err)
lastRunErr = err
// Deallocate this instance's resources
d.resourceMgr.deallocate(instanceType)
continue
@@ -789,8 +838,15 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
for range instances {
d.resourceMgr.deallocate(instanceType)
}
slog.Error("handleEC2RunInstances failed to create minimum instances", "created", len(instances), "minCount", minCount)
errResp = utils.GenerateErrorPayload(awserrors.ErrorServerInternal)
// Propagate the service-layer error if it's a known AWS error code
errCode := awserrors.ErrorServerInternal
if lastRunErr != nil {
if _, ok := awserrors.ErrorLookup[lastRunErr.Error()]; ok {
errCode = lastRunErr.Error()
}
}
slog.Error("handleEC2RunInstances failed to create minimum instances", "created", len(instances), "minCount", minCount, "err", errCode)
errResp = utils.GenerateErrorPayload(errCode)
if err := msg.Respond(errResp); err != nil {
slog.Error("Failed to respond to NATS request", "err", err)
}
@@ -1194,11 +1250,17 @@ func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) {
d.Instances.Mu.Lock()
defer d.Instances.Mu.Unlock()
// Filter instances if specific instance IDs were requested
// Validate and filter instances if specific instance IDs were requested
instanceIDFilter := make(map[string]bool)
if len(describeInstancesInput.InstanceIds) > 0 {
for _, id := range describeInstancesInput.InstanceIds {
if id != nil {
if id != nil && *id != "" {
if !strings.HasPrefix(*id, "i-") {
if err := msg.Respond(utils.GenerateErrorPayload(awserrors.ErrorInvalidInstanceIDMalformed)); err != nil {
slog.Error("Failed to respond to NATS request", "err", err)
}
return
}
instanceIDFilter[*id] = true
}
}
hive/daemon/daemon_handlers_test.go
+243 -3
View File
@@ -5,15 +5,19 @@ import (
"encoding/json"
"fmt"
"os"
"strings"
"testing"
"time"
"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/config"
handlers_ec2_account "github.com/mulgadc/hive/hive/handlers/ec2/account"
handlers_ec2_eigw "github.com/mulgadc/hive/hive/handlers/ec2/eigw"
handlers_ec2_image "github.com/mulgadc/hive/hive/handlers/ec2/image"
handlers_ec2_instance "github.com/mulgadc/hive/hive/handlers/ec2/instance"
handlers_ec2_key "github.com/mulgadc/hive/hive/handlers/ec2/key"
handlers_ec2_snapshot "github.com/mulgadc/hive/hive/handlers/ec2/snapshot"
handlers_ec2_tags "github.com/mulgadc/hive/hive/handlers/ec2/tags"
@@ -21,14 +25,15 @@ import (
"github.com/mulgadc/hive/hive/objectstore"
"github.com/mulgadc/hive/hive/qmp"
"github.com/mulgadc/hive/hive/vm"
"github.com/mulgadc/viperblock/viperblock"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// createFullTestDaemon creates a test daemon with ALL services initialized (including
// key, image, snapshot, tags, eigw, account) using in-memory object stores.
func createFullTestDaemon(t *testing.T, natsURL string) *Daemon {
// createFullTestDaemonWithStore creates a test daemon with ALL services initialized
// and returns the shared memory store for seeding test data.
func createFullTestDaemonWithStore(t *testing.T, natsURL string) (*Daemon, *objectstore.MemoryObjectStore) {
daemon := createTestDaemon(t, natsURL)
memStore := objectstore.NewMemoryObjectStore()
@@ -42,6 +47,13 @@ func createFullTestDaemon(t *testing.T, natsURL string) *Daemon {
daemon.eigwService = handlers_ec2_eigw.NewEgressOnlyIGWServiceImpl(cfg)
daemon.accountService = handlers_ec2_account.NewAccountSettingsServiceImpl(cfg)
return daemon, memStore
}
// createFullTestDaemon creates a test daemon with ALL services initialized (including
// key, image, snapshot, tags, eigw, account) using in-memory object stores.
func createFullTestDaemon(t *testing.T, natsURL string) *Daemon {
daemon, _ := createFullTestDaemonWithStore(t, natsURL)
return daemon
}
@@ -309,6 +321,164 @@ func TestHandleNodeDiscover(t *testing.T) {
assert.Equal(t, daemon.node, resp.Node)
}
// --- handleEC2RunInstances AMI validation tests ---
func TestHandleEC2RunInstances_InvalidAMI(t *testing.T) {
natsURL := sharedNATSURL
daemon := createFullTestDaemon(t, natsURL)
sub, err := daemon.natsConn.QueueSubscribe("ec2.RunInstances", "hive-workers", daemon.handleEC2RunInstances)
require.NoError(t, err)
defer sub.Unsubscribe()
input := &ec2.RunInstancesInput{
ImageId: aws.String("ami-nonexistent"),
InstanceType: aws.String(getTestInstanceType()),
MinCount: aws.Int64(1),
MaxCount: aws.Int64(1),
}
reqData, _ := json.Marshal(input)
reply, err := daemon.natsConn.Request("ec2.RunInstances", reqData, 5*time.Second)
require.NoError(t, err)
// Should return InvalidAMIID.NotFound, not ServerInternal
assert.Contains(t, string(reply.Data), "InvalidAMIID.NotFound")
}
func TestHandleEC2RunInstances_InvalidKeyPair(t *testing.T) {
natsURL := sharedNATSURL
daemon, memStore := createFullTestDaemonWithStore(t, natsURL)
// Seed a valid AMI so AMI validation passes
seedTestAMI(t, memStore, daemon.config.Predastore.Bucket, "ami-test123")
sub, err := daemon.natsConn.QueueSubscribe("ec2.RunInstances", "hive-workers", daemon.handleEC2RunInstances)
require.NoError(t, err)
defer sub.Unsubscribe()
input := &ec2.RunInstancesInput{
ImageId: aws.String("ami-test123"),
InstanceType: aws.String(getTestInstanceType()),
KeyName: aws.String("nonexistent-keypair"),
MinCount: aws.Int64(1),
MaxCount: aws.Int64(1),
}
reqData, _ := json.Marshal(input)
reply, err := daemon.natsConn.Request("ec2.RunInstances", reqData, 5*time.Second)
require.NoError(t, err)
// Should return InvalidKeyPair.NotFound, not proceed to launch
assert.Contains(t, string(reply.Data), "InvalidKeyPair.NotFound")
}
func TestHandleEC2RunInstances_ValidKeyPairPassesValidation(t *testing.T) {
natsURL := sharedNATSURL
daemon, memStore := createFullTestDaemonWithStore(t, natsURL)
// Seed a valid AMI
seedTestAMI(t, memStore, daemon.config.Predastore.Bucket, "ami-test456")
// Seed a valid key pair (public key + metadata)
bucket := daemon.config.Predastore.Bucket
_, err := memStore.PutObject(&awss3.PutObjectInput{
Bucket: aws.String(bucket),
Key: aws.String("keys/123456789/my-key"),
Body: strings.NewReader("ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAITest"),
})
require.NoError(t, err)
metadataJSON := `{"KeyPairId":"key-abc123","KeyName":"my-key","KeyFingerprint":"SHA256:test"}`
_, err = memStore.PutObject(&awss3.PutObjectInput{
Bucket: aws.String(bucket),
Key: aws.String("keys/123456789/key-abc123.json"),
Body: strings.NewReader(metadataJSON),
})
require.NoError(t, err)
sub, err := daemon.natsConn.QueueSubscribe("ec2.RunInstances", "hive-workers", daemon.handleEC2RunInstances)
require.NoError(t, err)
defer sub.Unsubscribe()
input := &ec2.RunInstancesInput{
ImageId: aws.String("ami-test456"),
InstanceType: aws.String(getTestInstanceType()),
KeyName: aws.String("my-key"),
MinCount: aws.Int64(1),
MaxCount: aws.Int64(1),
}
reqData, _ := json.Marshal(input)
reply, err := daemon.natsConn.Request("ec2.RunInstances", reqData, 5*time.Second)
require.NoError(t, err)
// Should NOT contain InvalidKeyPair.NotFound — key pair validation should pass
assert.NotContains(t, string(reply.Data), "InvalidKeyPair.NotFound")
}
func TestHandleEC2RunInstances_EmptyKeyNameSkipsValidation(t *testing.T) {
natsURL := sharedNATSURL
daemon, memStore := createFullTestDaemonWithStore(t, natsURL)
seedTestAMI(t, memStore, daemon.config.Predastore.Bucket, "ami-test789")
sub, err := daemon.natsConn.QueueSubscribe("ec2.RunInstances", "hive-workers", daemon.handleEC2RunInstances)
require.NoError(t, err)
defer sub.Unsubscribe()
// No KeyName at all — should skip validation
input := &ec2.RunInstancesInput{
ImageId: aws.String("ami-test789"),
InstanceType: aws.String(getTestInstanceType()),
MinCount: aws.Int64(1),
MaxCount: aws.Int64(1),
}
reqData, _ := json.Marshal(input)
reply, err := daemon.natsConn.Request("ec2.RunInstances", reqData, 5*time.Second)
require.NoError(t, err)
// Should NOT contain InvalidKeyPair.NotFound
assert.NotContains(t, string(reply.Data), "InvalidKeyPair.NotFound")
}
// --- handleEC2RunInstances service-layer error propagation ---
func TestHandleEC2RunInstances_ServiceErrorPropagated(t *testing.T) {
natsURL := sharedNATSURL
daemon, memStore := createFullTestDaemonWithStore(t, natsURL)
seedTestAMI(t, memStore, daemon.config.Predastore.Bucket, "ami-propatest")
// Override instanceService with one that has an empty instance types map.
// The resourceMgr still has instance types, so the daemon-level check passes,
// but RunInstance() will fail with ErrorInvalidInstanceType.
emptyTypes := map[string]*ec2.InstanceTypeInfo{}
daemon.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(
daemon.config, emptyTypes, daemon.natsConn, &daemon.Instances,
objectstore.NewMemoryObjectStore(),
)
sub, err := daemon.natsConn.QueueSubscribe("ec2.RunInstances", "hive-workers", daemon.handleEC2RunInstances)
require.NoError(t, err)
defer sub.Unsubscribe()
input := &ec2.RunInstancesInput{
ImageId: aws.String("ami-propatest"),
InstanceType: aws.String(getTestInstanceType()),
MinCount: aws.Int64(1),
MaxCount: aws.Int64(1),
}
reqData, _ := json.Marshal(input)
reply, err := daemon.natsConn.Request("ec2.RunInstances", reqData, 5*time.Second)
require.NoError(t, err)
// Should propagate the specific AWS error from the service layer,
// not swallow it into ServerInternal
assert.Contains(t, string(reply.Data), "InvalidInstanceType")
assert.NotContains(t, string(reply.Data), "ServerInternal")
}
// --- handleStopOrTerminateInstance tests (JetStream required for TransitionState) ---
func TestHandleEC2Events_StopInstance(t *testing.T) {
@@ -1204,3 +1374,73 @@ func TestHandleEC2GetConsoleOutput_NotFound(t *testing.T) {
// Should get an error response (instance not found)
assert.Contains(t, string(reply.Data), "InvalidInstanceID.NotFound")
}
// TestAttachVolume_ZoneMismatch verifies that attaching a volume in a different AZ
// returns InvalidVolume.ZoneMismatch instead of proceeding.
func TestAttachVolume_ZoneMismatch(t *testing.T) {
natsURL := sharedNATSURL
daemon, store := createFullTestDaemonWithStore(t, natsURL)
// Set the daemon's AZ
daemon.config.AZ = "us-east-1a"
instanceID := "i-test-az-mismatch"
volumeID := "vol-az-mismatch"
// Create a running instance
instance := &vm.VM{
ID: instanceID,
InstanceType: getTestInstanceType(),
Status: vm.StateRunning,
Instance: &ec2.Instance{},
QMPClient: &qmp.QMPClient{},
}
daemon.Instances.VMS[instanceID] = instance
// Create a volume in a different AZ
wrapper := struct {
VolumeConfig viperblock.VolumeConfig `json:"VolumeConfig"`
}{
VolumeConfig: viperblock.VolumeConfig{
VolumeMetadata: viperblock.VolumeMetadata{
VolumeID: volumeID,
SizeGiB: 10,
State: "available",
AvailabilityZone: "us-west-2a",
},
},
}
data, _ := json.Marshal(wrapper)
store.PutObject(&awss3.PutObjectInput{
Bucket: aws.String("test-bucket"),
Key: aws.String(volumeID + "/config.json"),
Body: strings.NewReader(string(data)),
})
// Subscribe handler
sub, err := daemon.natsConn.Subscribe(
fmt.Sprintf("ec2.cmd.%s", instanceID),
daemon.handleEC2Events,
)
require.NoError(t, err)
defer sub.Unsubscribe()
command := qmp.Command{
ID: instanceID,
Attributes: qmp.Attributes{
AttachVolume: true,
},
AttachVolumeData: &qmp.AttachVolumeData{
VolumeID: volumeID,
},
}
cmdData, _ := json.Marshal(command)
resp, err := daemon.natsConn.Request(
fmt.Sprintf("ec2.cmd.%s", instanceID),
cmdData,
5*time.Second,
)
require.NoError(t, err)
assert.Contains(t, string(resp.Data), "InvalidVolume.ZoneMismatch")
}
hive/daemon/daemon_test.go
+124 -1
View File
@@ -14,6 +14,7 @@ import (
"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/config"
handlers_ec2_instance "github.com/mulgadc/hive/hive/handlers/ec2/instance"
handlers_ec2_volume "github.com/mulgadc/hive/hive/handlers/ec2/volume"
@@ -103,6 +104,20 @@ func getTestInstanceType() string {
return "t3.micro" // Default fallback
}
// seedTestAMI creates a minimal AMI config in the memory store so that
// handleEC2RunInstances AMI validation passes.
func seedTestAMI(t *testing.T, store *objectstore.MemoryObjectStore, bucket, imageID string) {
t.Helper()
amiConfig := `{"VolumeConfig":{"AMIMetadata":{"ImageID":"` + imageID + `","Name":"test"}}}`
_, err := store.PutObject(&awss3.PutObjectInput{
Bucket: aws.String(bucket),
Key: aws.String(imageID + "/config.json"),
Body: strings.NewReader(amiConfig),
ContentType: aws.String("application/json"),
})
require.NoError(t, err)
}
// TestHandleEC2RunInstances_MessageParsing tests that the handler correctly parses NATS messages
func TestHandleEC2RunInstances_MessageParsing(t *testing.T) {
// Skip integration tests on macOS since viperblock/nbdkit are not available
@@ -1285,7 +1300,11 @@ func TestRunInstances_CountValidation(t *testing.T) {
instanceType := getTestInstanceType()
topic := fmt.Sprintf("ec2.RunInstances.%s", instanceType)
daemon := createTestDaemon(t, natsURL)
daemon, memStore := createFullTestDaemonWithStore(t, natsURL)
// Seed a valid AMI in the store so tests that pass input validation
// don't fail at the AMI existence check
seedTestAMI(t, memStore, daemon.config.Predastore.Bucket, "ami-test")
// Subscribe to the per-instance-type topic (matches production routing)
sub, err := daemon.natsConn.QueueSubscribe(topic, "hive-workers", daemon.handleEC2RunInstances)
@@ -3459,3 +3478,107 @@ func TestRollbackEBSMount_NATSTimeout(t *testing.T) {
// Should not panic, just log the timeout
daemon.rollbackEBSMount(ebsReq)
}
// TestDescribeInstances_InvalidInstanceIDMalformed verifies that DescribeInstances
// returns InvalidInstanceID.Malformed when given instance IDs without the i- prefix.
func TestDescribeInstances_InvalidInstanceIDMalformed(t *testing.T) {
natsURL := sharedNATSURL
daemon := createTestDaemon(t, natsURL)
sub, err := daemon.natsConn.Subscribe("ec2.DescribeInstances", daemon.handleEC2DescribeInstances)
require.NoError(t, err)
defer sub.Unsubscribe()
t.Run("MalformedInstanceID", func(t *testing.T) {
input := &ec2.DescribeInstancesInput{
InstanceIds: []*string{aws.String("bad-instance-id")},
}
inputJSON, _ := json.Marshal(input)
resp, err := daemon.natsConn.Request("ec2.DescribeInstances", inputJSON, 5*time.Second)
require.NoError(t, err)
assert.Contains(t, string(resp.Data), "InvalidInstanceID.Malformed")
})
t.Run("ValidInstanceIDPassesValidation", func(t *testing.T) {
input := &ec2.DescribeInstancesInput{
InstanceIds: []*string{aws.String("i-nonexistent")},
}
inputJSON, _ := json.Marshal(input)
resp, err := daemon.natsConn.Request("ec2.DescribeInstances", inputJSON, 5*time.Second)
require.NoError(t, err)
// Should not contain a malformed error — returns empty results instead
assert.NotContains(t, string(resp.Data), "InvalidInstanceID.Malformed")
})
}
// TestStopTerminate_IncorrectInstanceState verifies that stopping an already-stopped
// instance or terminating an already-terminated instance returns IncorrectInstanceState
// instead of ServerInternal.
func TestStopTerminate_IncorrectInstanceState(t *testing.T) {
natsURL := sharedNATSURL
daemon := createTestDaemon(t, natsURL)
instanceID := "i-test-state-check"
instance := &vm.VM{
ID: instanceID,
InstanceType: getTestInstanceType(),
Status: vm.StateStopped,
Instance: &ec2.Instance{},
}
daemon.Instances.VMS[instanceID] = instance
sub, err := daemon.natsConn.Subscribe(
fmt.Sprintf("ec2.cmd.%s", instanceID),
daemon.handleEC2Events,
)
require.NoError(t, err)
defer sub.Unsubscribe()
t.Run("StopAlreadyStoppedInstance", func(t *testing.T) {
daemon.Instances.Mu.Lock()
instance.Status = vm.StateStopped
daemon.Instances.Mu.Unlock()
command := qmp.Command{
ID: instanceID,
Attributes: qmp.Attributes{
StopInstance: true,
},
}
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")
assert.NotContains(t, string(resp.Data), "ServerInternal")
})
t.Run("TerminateAlreadyTerminatedInstance", func(t *testing.T) {
daemon.Instances.Mu.Lock()
instance.Status = vm.StateTerminated
daemon.Instances.Mu.Unlock()
command := qmp.Command{
ID: instanceID,
Attributes: qmp.Attributes{
TerminateInstance: true,
},
}
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")
assert.NotContains(t, string(resp.Data), "ServerInternal")
})
}
hive/handlers/ec2/image/service_impl.go
+69 -4
View File
@@ -225,6 +225,21 @@ func (s *ImageServiceImpl) DescribeImages(input *ec2.DescribeImagesInput) (*ec2.
images = append(images, image)
}
// If specific ImageIds were requested, verify all were found
if len(input.ImageIds) > 0 {
foundIDs := make(map[string]bool, len(images))
for _, img := range images {
if img.ImageId != nil {
foundIDs[*img.ImageId] = true
}
}
for _, reqID := range input.ImageIds {
if reqID != nil && !foundIDs[*reqID] {
return nil, errors.New(awserrors.ErrorInvalidAMIIDNotFound)
}
}
}
slog.Info("DescribeImages completed", "count", len(images))
return &ec2.DescribeImagesOutput{
@@ -246,6 +261,17 @@ func (s *ImageServiceImpl) CreateImageFromInstance(params CreateImageParams) (*e
return nil, errors.New(awserrors.ErrorInvalidParameterValue)
}
// Check for duplicate AMI name before doing any expensive work
name := aws.StringValue(input.Name)
if name != "" {
if exists, err := s.amiNameExists(name); err != nil {
slog.Error("CreateImageFromInstance: failed to check AMI name uniqueness", "name", name, "err", err)
return nil, errors.New(awserrors.ErrorServerInternal)
} else if exists {
return nil, errors.New(awserrors.ErrorInvalidAMINameDuplicate)
}
}
amiID := utils.GenerateResourceID("ami")
snapshotID := utils.GenerateResourceID("snap")
@@ -277,7 +303,7 @@ func (s *ImageServiceImpl) CreateImageFromInstance(params CreateImageParams) (*e
Virtualization: "hvm",
}
if params.SourceImageID != "" {
srcCfg, err := s.getAMIConfig(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)
@@ -292,7 +318,6 @@ func (s *ImageServiceImpl) CreateImageFromInstance(params CreateImageParams) (*e
}
// Step 5: Build and store AMI config
name := aws.StringValue(input.Name)
description := aws.StringValue(input.Description)
amiConfig := viperblock.VBState{
@@ -453,8 +478,48 @@ func (s *ImageServiceImpl) getVolumeConfig(volumeID string) (*viperblock.VolumeC
return &vbState.VolumeConfig, nil
}
// getAMIConfig reads an AMI's metadata from S3
func (s *ImageServiceImpl) getAMIConfig(imageID string) (viperblock.AMIMetadata, error) {
// amiNameExists checks if any existing AMI already uses the given name.
func (s *ImageServiceImpl) amiNameExists(name string) (bool, error) {
listResult, err := s.store.ListObjects(&s3.ListObjectsInput{
Bucket: aws.String(s.bucketName),
Prefix: aws.String("ami-"),
Delimiter: aws.String("/"),
})
if err != nil {
return false, fmt.Errorf("amiNameExists: failed to list AMIs: %w", err)
}
for _, prefix := range listResult.CommonPrefixes {
if prefix.Prefix == nil {
continue
}
configKey := *prefix.Prefix + "config.json"
result, err := s.store.GetObject(&s3.GetObjectInput{
Bucket: aws.String(s.bucketName),
Key: aws.String(configKey),
})
if err != nil {
continue
}
var vbState viperblock.VBState
decodeErr := json.NewDecoder(result.Body).Decode(&vbState)
_ = result.Body.Close()
if decodeErr != nil {
continue
}
if vbState.VolumeConfig.AMIMetadata.Name == name {
return true, nil
}
}
return false, nil
}
// GetAMIConfig retrieves the AMI metadata for a given image ID from S3.
// Returns NoSuchKeyError if the AMI does not exist.
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),
hive/handlers/ec2/image/service_impl_test.go
+106 -2
View File
@@ -74,6 +74,33 @@ func createTestAMIConfig(t *testing.T, store *objectstore.MemoryObjectStore, ima
require.NoError(t, err)
}
// createTestAMIConfigWithName creates a test AMI config with a specified name
func createTestAMIConfigWithName(t *testing.T, store *objectstore.MemoryObjectStore, imageID, name string) {
amiState := viperblock.VBState{
VolumeConfig: viperblock.VolumeConfig{
AMIMetadata: viperblock.AMIMetadata{
ImageID: imageID,
Name: name,
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)
@@ -162,7 +189,7 @@ func TestGetAMIConfig(t *testing.T) {
createTestAMIConfig(t, store, "ami-abc123")
meta, err := svc.getAMIConfig("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)
@@ -174,7 +201,7 @@ func TestGetAMIConfig(t *testing.T) {
func TestGetAMIConfig_NotFound(t *testing.T) {
svc, _ := setupTestImageService(t)
_, err := svc.getAMIConfig("ami-nonexistent")
_, err := svc.GetAMIConfig("ami-nonexistent")
require.Error(t, err)
}
@@ -223,6 +250,37 @@ func TestCreateImageFromInstance_SourceAMIReadFailure(t *testing.T) {
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
}
func TestDescribeImages_NotFound(t *testing.T) {
svc, store := setupTestImageService(t)
// Create one AMI
createTestAMIConfig(t, store, "ami-exists123")
// Request a non-existent AMI ID — should return InvalidAMIID.NotFound
_, err := svc.DescribeImages(&ec2.DescribeImagesInput{
ImageIds: []*string{aws.String("ami-nonexistent")},
})
require.Error(t, err)
assert.Equal(t, awserrors.ErrorInvalidAMIIDNotFound, err.Error())
}
func TestDescribeImages_MixedExistingAndMissing(t *testing.T) {
svc, store := setupTestImageService(t)
// Create one AMI
createTestAMIConfig(t, store, "ami-exists123")
// Request one existing + one non-existent — should return NotFound
_, err := svc.DescribeImages(&ec2.DescribeImagesInput{
ImageIds: []*string{
aws.String("ami-exists123"),
aws.String("ami-missing456"),
},
})
require.Error(t, err)
assert.Equal(t, awserrors.ErrorInvalidAMIIDNotFound, err.Error())
}
func TestDescribeImages_FilterByOwnerSelf(t *testing.T) {
svc, store := setupTestImageService(t)
@@ -259,3 +317,49 @@ func TestDescribeImages_FilterByOwnerSelf(t *testing.T) {
assert.Len(t, result.Images, 1)
assert.Equal(t, amiID, *result.Images[0].ImageId)
}
func TestCreateImageFromInstance_DuplicateName(t *testing.T) {
svc, store := setupTestImageService(t)
// Create an existing AMI with name "my-image"
createTestAMIConfigWithName(t, store, "ami-existing123", "my-image")
// Create volume config for the snapshot step
createTestVolumeConfig(t, store, "vol-root123", 10)
// Attempt to create another image with the same name — should fail with duplicate error
_, err := svc.CreateImageFromInstance(CreateImageParams{
Input: &ec2.CreateImageInput{
InstanceId: aws.String("i-test123"),
Name: aws.String("my-image"),
},
RootVolumeID: "vol-root123",
IsRunning: true,
})
require.Error(t, err)
assert.Equal(t, awserrors.ErrorInvalidAMINameDuplicate, err.Error())
}
func TestCreateImageFromInstance_UniqueNameAllowed(t *testing.T) {
svc, store := setupTestImageService(t)
// Create an existing AMI with a different name
createTestAMIConfigWithName(t, store, "ami-existing123", "other-image")
// Create volume config
createTestVolumeConfig(t, store, "vol-root123", 10)
// Creating with a unique name should NOT fail at the name check stage
// (it will fail later at the snapshot step since natsConn is nil, but that's expected)
_, err := svc.CreateImageFromInstance(CreateImageParams{
Input: &ec2.CreateImageInput{
InstanceId: aws.String("i-test123"),
Name: aws.String("unique-image"),
},
RootVolumeID: "vol-root123",
IsRunning: true,
})
require.Error(t, err)
// Should fail at snapshot, NOT at duplicate name check
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
}
hive/handlers/ec2/instance/service_impl.go
+6 -2
View File
@@ -318,8 +318,8 @@ func (s *InstanceServiceImpl) cloneAMIToVolume(input *ec2.RunInstancesInput, siz
amiState, err := amiVb.LoadStateRequest("")
if err != nil {
slog.Error("Could not load state for AMI", "err", err)
return errors.New(awserrors.ErrorServerInternal)
slog.Error("Could not load state for AMI", "imageId", *input.ImageId, "err", err)
return errors.New(awserrors.ErrorInvalidAMIIDNotFound)
}
snapshotID := amiState.VolumeConfig.AMIMetadata.SnapshotID
@@ -546,6 +546,10 @@ func (s *InstanceServiceImpl) createCloudInitISO(input *ec2.RunInstancesInput, i
Key: aws.String(keyPath),
})
if err != nil {
if objectstore.IsNoSuchKeyError(err) {
slog.Error("key pair not found", "keyName", keyName, "err", err)
return errors.New(awserrors.ErrorInvalidKeyPairNotFound)
}
slog.Error("failed to read SSH key", "err", err)
return errors.New(awserrors.ErrorServerInternal)
}
hive/handlers/ec2/key/service_impl.go
+25 -2
View File
@@ -362,6 +362,13 @@ func (s *KeyServiceImpl) findKeyPairIdFromKeyName(keyName string) (string, error
return "", errors.New(awserrors.ErrorInvalidKeyPairNotFound)
}
// ValidateKeyPairExists checks if a key pair with the given name exists.
// Returns nil if the key pair exists, or an error with ErrorInvalidKeyPairNotFound if not.
func (s *KeyServiceImpl) ValidateKeyPairExists(keyName string) error {
_, err := s.findKeyPairIdFromKeyName(keyName)
return err
}
// DeleteKeyPair removes a key pair (both public key and metadata from S3)
func (s *KeyServiceImpl) DeleteKeyPair(input *ec2.DeleteKeyPairInput) (*ec2.DeleteKeyPairOutput, error) {
if input == nil {
@@ -386,6 +393,11 @@ func (s *KeyServiceImpl) DeleteKeyPair(input *ec2.DeleteKeyPairInput) (*ec2.Dele
keyName, err = s.getKeyNameFromKeyPairId(keyPairID)
if err != nil {
// AWS DeleteKeyPair is idempotent — return success for non-existent keys
if err.Error() == awserrors.ErrorInvalidKeyPairNotFound {
slog.Debug("DeleteKeyPair: key pair not found, returning success (idempotent)", "keyPairId", keyPairID)
return &ec2.DeleteKeyPairOutput{}, nil
}
slog.Error("Failed to get keyName from keyPairId", "keyPairId", keyPairID, "err", err)
return nil, err
}
@@ -401,6 +413,11 @@ func (s *KeyServiceImpl) DeleteKeyPair(input *ec2.DeleteKeyPairInput) (*ec2.Dele
keyPairID, err = s.findKeyPairIdFromKeyName(keyName)
if err != nil {
// AWS DeleteKeyPair is idempotent — return success for non-existent keys
if err.Error() == awserrors.ErrorInvalidKeyPairNotFound {
slog.Debug("DeleteKeyPair: key pair not found, returning success (idempotent)", "keyName", keyName)
return &ec2.DeleteKeyPairOutput{}, nil
}
slog.Error("Failed to find keyPairId from keyName", "keyName", keyName, "err", err)
return nil, err
}
@@ -596,7 +613,7 @@ func (s *KeyServiceImpl) ImportKeyPair(input *ec2.ImportKeyPairInput) (*ec2.Impo
parts := strings.Fields(publicKeyString)
if len(parts) < 2 {
slog.Error("Invalid public key format", "keyName", keyName)
return nil, errors.New(awserrors.ErrorInvalidKeyPairFormat)
return nil, errors.New(awserrors.ErrorInvalidKeyFormat)
}
// Determine key type from algorithm prefix
@@ -612,7 +629,13 @@ func (s *KeyServiceImpl) ImportKeyPair(input *ec2.ImportKeyPairInput) (*ec2.Impo
keyType = "ecdsa"
default:
slog.Error("Unsupported key type", "algorithm", algorithmPrefix, "keyName", keyName)
return nil, errors.New(awserrors.ErrorInvalidKeyPairFormat)
return nil, errors.New(awserrors.ErrorInvalidKeyFormat)
}
// Validate that the key data is valid base64
if _, err := base64.StdEncoding.DecodeString(parts[1]); err != nil {
slog.Error("Invalid base64 in public key material", "keyName", keyName, "err", err)
return nil, errors.New(awserrors.ErrorInvalidKeyFormat)
}
// Calculate fingerprint from the imported public key
hive/handlers/ec2/key/service_impl_test.go
+71
View File
@@ -7,6 +7,9 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/hive/hive/objectstore"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// TestKeyPairMetadataMarshaling tests that CreateKeyPairOutput (used as metadata) can be marshaled/unmarshaled correctly
@@ -389,6 +392,74 @@ func TestImportKeyPairKeyTypeDetection(t *testing.T) {
}
}
// TestDeleteKeyPairIdempotent verifies AWS-compatible idempotent behavior:
// deleting a non-existent key pair returns success instead of NotFound.
func TestDeleteKeyPairIdempotent(t *testing.T) {
store := objectstore.NewMemoryObjectStore()
svc := NewKeyServiceImplWithStore(store, "test-bucket", "123456789")
t.Run("NonExistentKeyName", func(t *testing.T) {
result, err := svc.DeleteKeyPair(&ec2.DeleteKeyPairInput{
KeyName: aws.String("no-such-key"),
})
require.NoError(t, err)
assert.NotNil(t, result)
})
t.Run("NonExistentKeyPairId", func(t *testing.T) {
result, err := svc.DeleteKeyPair(&ec2.DeleteKeyPairInput{
KeyPairId: aws.String("key-0123456789abcdef0"),
})
require.NoError(t, err)
assert.NotNil(t, result)
})
}
// TestImportKeyPairInvalidKeyFormat verifies that ImportKeyPair returns
// InvalidKey.Format for malformed public key material (not InvalidKeyPair.Format).
func TestImportKeyPairInvalidKeyFormat(t *testing.T) {
store := objectstore.NewMemoryObjectStore()
svc := NewKeyServiceImplWithStore(store, "test-bucket", "123456789")
tests := []struct {
name string
publicKey string
expectedErrMsg string
}{
{
name: "SingleFieldNoKeyData",
publicKey: "ssh-rsa",
expectedErrMsg: "InvalidKey.Format",
},
{
name: "UnsupportedAlgorithm",
publicKey: "ssh-dss AAAAB3NzaC1kc3MAAACB",
expectedErrMsg: "InvalidKey.Format",
},
{
name: "InvalidBase64",
publicKey: "ssh-rsa not-valid-base64!!!",
expectedErrMsg: "InvalidKey.Format",
},
{
name: "EmptyKeyData",
publicKey: "ssh-ed25519 ",
expectedErrMsg: "InvalidKey.Format",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
_, err := svc.ImportKeyPair(&ec2.ImportKeyPairInput{
KeyName: aws.String("test-key"),
PublicKeyMaterial: []byte(tt.publicKey),
})
require.Error(t, err)
assert.Equal(t, tt.expectedErrMsg, err.Error())
})
}
}
// TestImportKeyPairInputValidation tests input validation for ImportKeyPair
func TestImportKeyPairInputValidation(t *testing.T) {
tests := []struct {
hive/handlers/ec2/snapshot/service_impl.go
+52
View File
@@ -354,6 +354,47 @@ func (s *SnapshotServiceImpl) DescribeSnapshots(input *ec2.DescribeSnapshotsInpu
}, nil
}
// snapshotInUseByVolumes checks if any volume was created from the given snapshot.
func (s *SnapshotServiceImpl) snapshotInUseByVolumes(snapshotID string) (bool, error) {
listResult, err := s.store.ListObjects(&s3.ListObjectsInput{
Bucket: aws.String(s.config.Predastore.Bucket),
Prefix: aws.String("vol-"),
Delimiter: aws.String("/"),
})
if err != nil {
return false, fmt.Errorf("snapshotInUseByVolumes: failed to list volumes: %w", err)
}
for _, prefix := range listResult.CommonPrefixes {
if prefix.Prefix == nil {
continue
}
volumeID := strings.TrimSuffix(*prefix.Prefix, "/")
configKey := fmt.Sprintf("%s/config.json", volumeID)
result, err := s.store.GetObject(&s3.GetObjectInput{
Bucket: aws.String(s.config.Predastore.Bucket),
Key: aws.String(configKey),
})
if err != nil {
continue // volume may not have a config yet
}
var state viperblock.VBState
decodeErr := json.NewDecoder(result.Body).Decode(&state)
_ = result.Body.Close()
if decodeErr != nil {
continue
}
if state.VolumeConfig.VolumeMetadata.SnapshotID == snapshotID {
return true, nil
}
}
return false, nil
}
// DeleteSnapshot deletes a snapshot
func (s *SnapshotServiceImpl) DeleteSnapshot(input *ec2.DeleteSnapshotInput) (*ec2.DeleteSnapshotOutput, error) {
if input == nil || input.SnapshotId == nil {
@@ -370,6 +411,17 @@ func (s *SnapshotServiceImpl) DeleteSnapshot(input *ec2.DeleteSnapshotInput) (*e
return nil, err
}
// Check if any volumes were created from this snapshot
inUse, err := s.snapshotInUseByVolumes(snapshotID)
if err != nil {
slog.Error("DeleteSnapshot failed to check snapshot usage", "snapshotId", snapshotID, "err", err)
return nil, errors.New(awserrors.ErrorServerInternal)
}
if inUse {
slog.Info("DeleteSnapshot blocked: snapshot in use by volume", "snapshotId", snapshotID)
return nil, errors.New(awserrors.ErrorInvalidSnapshotInUse)
}
listResult, err := s.store.ListObjects(&s3.ListObjectsInput{
Bucket: aws.String(s.config.Predastore.Bucket),
Prefix: aws.String(snapshotID + "/"),
hive/handlers/ec2/snapshot/service_impl_test.go
+44
View File
@@ -249,6 +249,50 @@ func TestDeleteSnapshot(t *testing.T) {
assert.Empty(t, result.Snapshots)
}
// TestDeleteSnapshot_InUseByVolume tests that deleting a snapshot fails when a volume was created from it
func TestDeleteSnapshot_InUseByVolume(t *testing.T) {
svc, store := setupTestSnapshotService(t)
// Create a test volume and snapshot
createTestVolume(t, store, "vol-source", 50)
snap, err := svc.CreateSnapshot(&ec2.CreateSnapshotInput{
VolumeId: aws.String("vol-source"),
})
require.NoError(t, err)
// Create a volume that references this snapshot (simulates CreateVolume from snapshot)
volumeState := viperblock.VBState{
VolumeConfig: viperblock.VolumeConfig{
VolumeMetadata: viperblock.VolumeMetadata{
SizeGiB: 50,
SnapshotID: *snap.SnapshotId,
},
},
}
data, err := json.Marshal(volumeState)
require.NoError(t, err)
_, err = store.PutObject(&s3.PutObjectInput{
Bucket: aws.String("test-bucket"),
Key: aws.String("vol-cloned/config.json"),
Body: strings.NewReader(string(data)),
})
require.NoError(t, err)
// Attempt to delete the snapshot — should fail
_, err = svc.DeleteSnapshot(&ec2.DeleteSnapshotInput{
SnapshotId: snap.SnapshotId,
})
require.Error(t, err)
assert.Contains(t, err.Error(), awserrors.ErrorInvalidSnapshotInUse)
// Verify snapshot still exists
result, err := svc.DescribeSnapshots(&ec2.DescribeSnapshotsInput{
SnapshotIds: []*string{snap.SnapshotId},
})
require.NoError(t, err)
assert.Len(t, result.Snapshots, 1)
}
// TestDeleteSnapshot_NotFound tests deleting a non-existent snapshot
func TestDeleteSnapshot_NotFound(t *testing.T) {
svc, _ := setupTestSnapshotService(t)
hive/handlers/ec2/volume/service_impl.go
+11
View File
@@ -234,7 +234,18 @@ func (s *VolumeServiceImpl) DescribeVolumes(input *ec2.DescribeVolumesInput) (*e
// Fast path: if specific volume IDs are requested, fetch them directly
if len(input.VolumeIds) > 0 {
// Count non-nil requested IDs
requested := 0
for _, id := range input.VolumeIds {
if id != nil {
requested++
}
}
volumes = s.fetchVolumesByIDs(input.VolumeIds)
// AWS returns InvalidVolume.NotFound if any requested ID is missing
if len(volumes) != requested {
return nil, errors.New(awserrors.ErrorInvalidVolumeNotFound)
}
slog.Info("DescribeVolumes completed", "count", len(volumes))
return &ec2.DescribeVolumesOutput{Volumes: volumes}, nil
}
hive/handlers/ec2/volume/service_impl_test.go
+4 -4
View File
@@ -784,12 +784,12 @@ func TestDescribeVolumes_FastPath_MixedExistingAndMissing(t *testing.T) {
VolumeID: "vol-exists", SizeGiB: 10, State: "available",
})
output, err := svc.DescribeVolumes(&ec2.DescribeVolumesInput{
// AWS returns InvalidVolume.NotFound when any requested ID is missing
_, err := svc.DescribeVolumes(&ec2.DescribeVolumesInput{
VolumeIds: []*string{aws.String("vol-exists"), aws.String("vol-ghost")},
})
require.NoError(t, err)
assert.Len(t, output.Volumes, 1)
assert.Equal(t, "vol-exists", *output.Volumes[0].VolumeId)
require.Error(t, err)
assert.Equal(t, awserrors.ErrorInvalidVolumeNotFound, err.Error())
}
func TestDescribeVolumes_FastPath_NilVolumeID(t *testing.T) {