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

Merge commit '13d9fb494a699a03acde4ab02d2bc78152638734' into feat/ddil-scenarios-abc

Browse Source
This commit is contained in:
joshsiv-mulga
2026-05-11 13:53:27 +10:00
parent 0cf3627430 13d9fb494a
commit 381afd440d
15 changed files with 2395 additions and 1332 deletions
Download Patch File Download Diff File Expand all files Collapse all files
spinifex/daemon/daemon.go
+28
View File
@@ -1096,6 +1096,19 @@ func (d *Daemon) startCluster() error {
}
}
// Wire deps needed by InstanceService.TerminateStoppedInstance now that
// volumeService / vpcService / externalIPAM are constructed.
d.instanceService.SetTerminationDeps(d.volumeService, d.vpcService, d.externalIPAM)
// Wire deps for InstanceService.PrepareRunInstances (AMI/key/VPC/IPAM).
d.instanceService.SetRunInstancesDeps(d.imageService, d.keyService, &daemonENICreator{d: d}, d.externalIPAM)
// Wire GPU claimer for InstanceService.StartStoppedInstance — nil when no
// passthrough hardware is configured.
if d.gpuManager != nil {
d.instanceService.SetGPUClaimer(&daemonGPUClaimer{d: d})
}
d.accountService, err = initServiceWithRetry("account settings service", func() (*handlers_ec2_account.AccountSettingsServiceImpl, error) {
return handlers_ec2_account.NewAccountSettingsServiceImplWithNATS(d.config, d.natsConn)
})
@@ -1844,6 +1857,21 @@ func (rm *ResourceManager) deallocate(instanceType *ec2.InstanceTypeInfo) {
rm.updateInstanceSubscriptions()
}
// Allocate, Deallocate, CanAllocate, InstanceTypes are the exported facade
// satisfying handlers_ec2_instance.InstanceTypeAllocator. The unexported
// variants stay for internal daemon callers.
func (rm *ResourceManager) Allocate(it *ec2.InstanceTypeInfo) error { return rm.allocate(it) }
func (rm *ResourceManager) Deallocate(it *ec2.InstanceTypeInfo) { rm.deallocate(it) }
func (rm *ResourceManager) CanAllocate(it *ec2.InstanceTypeInfo, count int) int {
return rm.canAllocate(it, count)
}
// InstanceTypes returns the shared instance-type map. Callers must not mutate
// the returned map; reloadGPUTypes mutates it in place under rm.mu.
func (rm *ResourceManager) InstanceTypes() map[string]*ec2.InstanceTypeInfo {
return rm.instanceTypes
}
// initSubscriptions sets up dynamic per-instance-type NATS subscriptions.
// Called once during daemon startup after NATS is connected.
// reloadGPUTypes replaces GPU instance types in-place and updates NATS subscriptions.
spinifex/daemon/daemon_handlers.go
+21 -3
View File
@@ -87,11 +87,29 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) {
case command.Attributes.DetachVolume:
d.handleDetachVolume(msg, command, instance)
case command.Attributes.StartInstance:
d.handleStartInstance(msg, command, instance)
if err := d.instanceService.StartInstance(instance, command); err != nil {
respondWithError(msg, awserrors.ValidErrorCode(err.Error()))
return
}
if err := msg.Respond(fmt.Appendf(nil, `{"status":"running","instanceId":"%s"}`, instance.ID)); err != nil {
slog.Error("Failed to respond to NATS request", "err", err)
}
case command.Attributes.RebootInstance:
d.handleRebootInstance(msg, command, instance)
if err := d.instanceService.RebootInstance(instance, command); err != nil {
respondWithError(msg, awserrors.ValidErrorCode(err.Error()))
return
}
if err := msg.Respond([]byte(`{}`)); err != nil {
slog.Error("Failed to respond to NATS request", "err", err)
}
case command.Attributes.StopInstance, command.Attributes.TerminateInstance:
d.handleStopOrTerminateInstance(msg, command, instance)
if err := d.instanceService.StopOrTerminateInstance(instance, command); err != nil {
respondWithError(msg, awserrors.ValidErrorCode(err.Error()))
return
}
if err := msg.Respond([]byte(`{}`)); err != nil {
slog.Error("Failed to respond to NATS request", "err", err)
}
default:
slog.Warn("Unhandled EC2 instance command", "id", command.ID, "attributes", command.Attributes)
respondWithError(msg, awserrors.ErrorServerInternal)
spinifex/daemon/daemon_handlers_instance.go
+18 -873
View File
@@ -1,31 +1,29 @@
package daemon
import (
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"log/slog"
"strings"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/awserrors"
"github.com/mulgadc/spinifex/spinifex/filterutil"
"github.com/mulgadc/spinifex/spinifex/instancetypes"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/vm"
"github.com/nats-io/nats.go"
)
// handleEC2RunInstances processes incoming EC2 RunInstances requests
// handleEC2RunInstances orchestrates the RunInstances flow across
// InstanceService.PrepareRunInstances (validation + ENI creation),
// daemon-local Insert/WriteState/per-instance subscribe, and
// InstanceService.LaunchRunInstances (volumes + GPU + vmMgr.Run). The split
// preserves the original respond-then-launch timing — AWS gets a reservation
// before the launch loop starts.
func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
slog.Debug("Received message on subject", "subject", msg.Subject)
slog.Debug("Message data", "data", string(msg.Data))
// Extract account ID from NATS header
accountID := utils.AccountIDFromMsg(msg)
if accountID == "" {
slog.Error("handleEC2RunInstances: missing account ID in NATS header")
@@ -33,11 +31,8 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
return
}
// Initialize runInstancesInput before unmarshaling into it
runInstancesInput := &ec2.RunInstancesInput{}
errResp := utils.UnmarshalJsonPayload(runInstancesInput, msg.Data)
if errResp != nil {
input := &ec2.RunInstancesInput{}
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)
}
@@ -45,271 +40,23 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
return
}
slog.Info("Processing RunInstances request for instance type", "instanceType", *runInstancesInput.InstanceType)
// Check if instance type is supported
instanceType, exists := d.resourceMgr.instanceTypes[*runInstancesInput.InstanceType]
if !exists {
slog.Error("handleEC2RunInstances instance lookup", "err", awserrors.ErrorInvalidInstanceType, "InstanceType", *runInstancesInput.InstanceType)
respondWithError(msg, awserrors.ErrorInvalidInstanceType)
return
}
// Validate AMI exists before allocating resources
if runInstancesInput.ImageId == nil || *runInstancesInput.ImageId == "" {
slog.Error("handleEC2RunInstances missing ImageId")
respondWithError(msg, awserrors.ErrorMissingParameter)
return
}
if d.imageService == nil {
slog.Error("handleEC2RunInstances image service not initialized")
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
amiMeta, err := d.imageService.GetAMIConfig(*runInstancesInput.ImageId)
reservation, instances, instanceType, err := d.instanceService.PrepareRunInstances(input, accountID)
if err != nil {
slog.Error("handleEC2RunInstances AMI not found", "imageId", *runInstancesInput.ImageId, "err", err)
respondWithError(msg, awserrors.ErrorInvalidAMIIDNotFound)
return
}
// Verify the caller can use this AMI: must own it or the AMI must
// have a non-account-ID owner alias (e.g. "self", "spinifex", empty)
// indicating a system/legacy AMI.
amiOwner := amiMeta.ImageOwnerAlias
if amiOwner != "" && amiOwner != accountID {
if utils.IsAccountID(amiOwner) {
slog.Warn("handleEC2RunInstances AMI not owned by caller", "imageId", *runInstancesInput.ImageId, "amiOwner", amiOwner, "accountID", accountID)
respondWithError(msg, awserrors.ErrorInvalidAMIIDNotFound)
return
}
}
// Validate key pair exists (if specified)
if runInstancesInput.KeyName != nil && *runInstancesInput.KeyName != "" {
if err := d.keyService.ValidateKeyPairExists(accountID, *runInstancesInput.KeyName); err != nil {
slog.Error("handleEC2RunInstances key pair not found", "keyName", *runInstancesInput.KeyName, "err", err)
respondWithError(msg, awserrors.ErrorInvalidKeyPairNotFound)
return
}
}
// Determine how many instances to launch based on MinCount/MaxCount
minCount := int(*runInstancesInput.MinCount)
maxCount := int(*runInstancesInput.MaxCount)
// Check how many we can actually launch
allocatableCount := d.resourceMgr.canAllocate(instanceType, maxCount)
if allocatableCount < minCount {
// Cannot satisfy MinCount requirement - fail entirely
slog.Error("handleEC2RunInstances insufficient capacity", "requested", minCount, "available", allocatableCount, "InstanceType", *runInstancesInput.InstanceType)
respondWithError(msg, awserrors.ErrorInsufficientInstanceCapacity)
respondWithError(msg, awserrors.ValidErrorCode(err.Error()))
return
}
// Launch up to MaxCount, capped by available capacity
// Note: canAllocate() already caps at maxCount, so allocatableCount <= maxCount
launchCount := allocatableCount
slog.Info("Instance count determined", "min", minCount, "max", maxCount, "launching", launchCount)
// Allocate resources for all instances upfront
var allocatedCount int
for i := 0; i < launchCount; i++ {
if err := d.resourceMgr.allocate(instanceType); err != nil {
slog.Error("handleEC2RunInstances allocate failed mid-allocation", "allocated", allocatedCount, "err", err)
break
}
allocatedCount++
}
// Check if we still meet MinCount after allocation
if allocatedCount < minCount {
// Rollback allocations
for i := 0; i < allocatedCount; i++ {
d.resourceMgr.deallocate(instanceType)
}
slog.Error("handleEC2RunInstances insufficient capacity after allocation", "allocated", allocatedCount, "minCount", minCount)
respondWithError(msg, awserrors.ErrorInsufficientInstanceCapacity)
return
}
// Update launchCount to what we actually allocated
launchCount = allocatedCount
// Delegate to service for business logic (volume creation, cloud-init, etc.)
instanceTypeName := ""
if instanceType.InstanceType != nil {
instanceTypeName = *instanceType.InstanceType
}
slog.Info("Launching EC2 instances", "instanceType", instanceTypeName, "count", launchCount)
// 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
}
// When Terraform sets associate_public_ip_address, it sends the subnet
// and security groups inside NetworkInterfaces[0] instead of the top-level
// fields. Extract them so the rest of the handler works uniformly.
if (runInstancesInput.SubnetId == nil || *runInstancesInput.SubnetId == "") &&
len(runInstancesInput.NetworkInterfaces) > 0 && runInstancesInput.NetworkInterfaces[0] != nil {
nic := runInstancesInput.NetworkInterfaces[0]
if nic.SubnetId != nil && *nic.SubnetId != "" {
runInstancesInput.SubnetId = nic.SubnetId
}
if len(runInstancesInput.SecurityGroupIds) == 0 && len(nic.Groups) > 0 {
runInstancesInput.SecurityGroupIds = nic.Groups
}
}
// Resolve default subnet when none specified (matches AWS behavior)
if (runInstancesInput.SubnetId == nil || *runInstancesInput.SubnetId == "") && d.vpcService != nil {
defaultSubnet, dsErr := d.vpcService.GetDefaultSubnet(accountID)
if dsErr == nil {
runInstancesInput.SubnetId = aws.String(defaultSubnet.SubnetId)
slog.Info("Resolved default subnet for instance", "instanceId", instance.ID, "subnetId", defaultSubnet.SubnetId)
}
}
// Auto-create ENI when SubnetId is provided (matches AWS behavior)
if runInstancesInput.SubnetId != nil && *runInstancesInput.SubnetId != "" && d.vpcService != nil {
eniOut, eniErr := d.vpcService.CreateNetworkInterface(&ec2.CreateNetworkInterfaceInput{
SubnetId: runInstancesInput.SubnetId,
Description: aws.String("Primary network interface for " + instance.ID),
}, accountID)
if eniErr != nil {
slog.Error("handleEC2RunInstances auto-create ENI failed", "instanceId", instance.ID, "subnetId", *runInstancesInput.SubnetId, "err", eniErr)
lastRunErr = eniErr
d.resourceMgr.deallocate(instanceType)
continue
}
eni := eniOut.NetworkInterface
instance.ENIId = *eni.NetworkInterfaceId
instance.ENIMac = *eni.MacAddress
// Mark ENI as attached to this instance so attachment.instance-id
// filter works (used by ELBv2 RegisterTargets to resolve target IPs).
if _, attachErr := d.vpcService.AttachENI(accountID, instance.ENIId, instance.ID, 0); attachErr != nil {
slog.Error("Failed to attach ENI to instance record — ELBv2 target IP resolution will fail", "eniId", instance.ENIId, "instanceId", instance.ID, "err", attachErr)
}
ec2Instance.SetPrivateIpAddress(*eni.PrivateIpAddress)
ec2Instance.SetSubnetId(*runInstancesInput.SubnetId)
ec2Instance.SetVpcId(*eni.VpcId)
ec2Instance.NetworkInterfaces = []*ec2.InstanceNetworkInterface{
{
NetworkInterfaceId: eni.NetworkInterfaceId,
PrivateIpAddress: eni.PrivateIpAddress,
MacAddress: eni.MacAddress,
SubnetId: runInstancesInput.SubnetId,
VpcId: eni.VpcId,
Status: aws.String("in-use"),
Attachment: &ec2.InstanceNetworkInterfaceAttachment{
DeviceIndex: aws.Int64(0),
Status: aws.String("attached"),
},
},
}
slog.Info("Auto-created ENI for VPC instance",
"instanceId", instance.ID,
"eniId", instance.ENIId,
"privateIp", *eni.PrivateIpAddress,
"mac", instance.ENIMac,
)
// Auto-assign public IP if subnet has MapPublicIpOnLaunch and external IPAM is available
if d.externalIPAM != nil {
subnet, subErr := d.vpcService.GetSubnet(accountID, *runInstancesInput.SubnetId)
if subErr == nil && subnet.MapPublicIpOnLaunch {
region := ""
az := ""
if d.config != nil {
region = d.config.Region
az = d.config.AZ
}
publicIP, poolName, allocErr := d.externalIPAM.AllocateIP(region, az, "auto_assign", "", *eni.NetworkInterfaceId, instance.ID)
if allocErr != nil {
slog.Warn("Failed to allocate public IP for instance", "instanceId", instance.ID, "err", allocErr)
} else {
// Update ENI record with public IP
if updateErr := d.vpcService.UpdateENIPublicIP(accountID, *eni.NetworkInterfaceId, publicIP, poolName); updateErr != nil {
slog.Warn("Failed to update ENI with public IP", "eniId", *eni.NetworkInterfaceId, "err", updateErr)
}
// Publish vpc.add-nat for dnat_and_snat rule
portName := "port-" + *eni.NetworkInterfaceId
d.publishNATEvent("vpc.add-nat", *eni.VpcId, publicIP, *eni.PrivateIpAddress, portName, *eni.MacAddress)
// Set on ec2Instance response
ec2Instance.PublicIpAddress = aws.String(publicIP)
instance.PublicIP = publicIP
instance.PublicIPPool = poolName
slog.Info("Auto-assigned public IP",
"instanceId", instance.ID,
"publicIp", publicIP,
"privateIp", *eni.PrivateIpAddress,
"pool", poolName,
)
}
}
}
}
instances = append(instances, instance)
allEC2Instances = append(allEC2Instances, ec2Instance)
}
// Check if we still have enough instances after creation errors
if len(instances) < minCount {
// Rollback: deallocate resources for successfully created instances
// (failed instances already deallocated their resources above)
for range instances {
d.resourceMgr.deallocate(instanceType)
}
// 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)
respondWithError(msg, errCode)
return
}
// Build reservation with all instances
reservation := ec2.Reservation{}
reservation.SetReservationId(utils.GenerateResourceID("r"))
reservation.SetOwnerId(accountID)
reservation.Instances = allEC2Instances
// Store reservation reference, account ID, and placement group in all VMs
// PlacementGroupNode is daemon-local identity, set after prepare.
for _, instance := range instances {
instance.Reservation = &reservation
instance.AccountID = accountID
if runInstancesInput.Placement != nil && runInstancesInput.Placement.GroupName != nil && *runInstancesInput.Placement.GroupName != "" {
instance.PlacementGroupName = *runInstancesInput.Placement.GroupName
if instance.PlacementGroupName != "" {
instance.PlacementGroupNode = d.node
}
}
// Respond to NATS immediately with reservation (instances are provisioning)
jsonResponse, err := json.Marshal(reservation)
if err != nil {
slog.Error("handleEC2RunInstances failed to marshal reservation", "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
// Deallocate all resources
for range instances {
d.resourceMgr.deallocate(instanceType)
}
@@ -319,21 +66,16 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
slog.Error("Failed to respond to NATS request", "err", err)
}
// Add all instances to state immediately so DescribeInstances can find them
// while volumes are being prepared and VMs are launching
for _, instance := range instances {
d.vmMgr.Insert(instance)
}
if err := d.WriteState(); err != nil {
slog.Error("handleEC2RunInstances failed to write initial state", "err", err)
}
slog.Info("Instances added to state with pending status", "count", len(instances))
// Subscribe to per-instance NATS topics early so terminate/stop commands
// can reach this daemon while volumes are being prepared. LaunchInstance
// will replace these subscriptions when it completes.
// Subscribe per-instance NATS topics so terminate/stop reach this daemon
// while volumes prepare. LaunchInstance replaces these on success.
d.mu.Lock()
for _, instance := range instances {
sub, subErr := d.natsConn.Subscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), d.handleEC2Events)
@@ -345,214 +87,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) {
}
d.mu.Unlock()
// Launch all instances (volumes and VMs)
var successCount int
for _, instance := range instances {
// Skip if instance was terminated by a concurrent request
status := d.vmMgr.Status(instance)
if status != vm.StatePending && status != vm.StateProvisioning {
slog.Info("Instance state changed during provisioning, skipping launch",
"instanceId", instance.ID, "status", string(status))
continue
}
// Pre-compute dev MAC so cloud-init can generate per-interface netplan
// that suppresses the default route on the dev/hostfwd NIC.
if d.config.Daemon.DevNetworking && instance.ENIId != "" {
instance.DevMAC = vm.GenerateDevMAC(instance.ID)
}
// Prepare the root volume, cloud-init, EFI drives via NBD (AMI clone to new volume)
volumeInfos, err := d.instanceService.GenerateVolumes(runInstancesInput, instance)
if err != nil {
slog.Error("handleEC2RunInstances GenerateVolumes failed", "instanceId", instance.ID, "err", err)
d.vmMgr.MarkFailed(instance, "volume_preparation_failed")
continue
}
// Populate BlockDeviceMappings on the ec2.Instance
instance.Instance.BlockDeviceMappings = make([]*ec2.InstanceBlockDeviceMapping, 0, len(volumeInfos))
for _, vi := range volumeInfos {
mapping := &ec2.InstanceBlockDeviceMapping{}
mapping.SetDeviceName(vi.DeviceName)
mapping.Ebs = &ec2.EbsInstanceBlockDevice{}
mapping.Ebs.SetVolumeId(vi.VolumeId)
mapping.Ebs.SetAttachTime(vi.AttachTime)
mapping.Ebs.SetDeleteOnTermination(vi.DeleteOnTermination)
mapping.Ebs.SetStatus("attached")
instance.Instance.BlockDeviceMappings = append(instance.Instance.BlockDeviceMappings, mapping)
}
// Claim GPU for GPU instance types — binds the full IOMMU group to vfio-pci.
if d.gpuManager != nil && instancetypes.IsGPUType(instanceType) {
gpuDevice, gpuErr := d.gpuManager.Claim(instance.ID)
if gpuErr != nil {
slog.Error("handleEC2RunInstances: GPU claim failed", "instanceId", instance.ID, "err", gpuErr)
d.vmMgr.MarkFailed(instance, "gpu_claim_failed")
continue
}
instance.GPUPCIAddress = gpuDevice.PCIAddress
instance.GPUXVGAEnabled = gpuXVGAEnabled(gpuDevice, d.config.Daemon.GPUModelOverrides)
slog.Info("GPU claimed for instance", "instanceId", instance.ID, "gpu", gpuDevice.PCIAddress, "xvga", instance.GPUXVGAEnabled)
}
// Launch the instance infrastructure (QEMU, QMP, NATS subscriptions)
err = d.vmMgr.Run(instance)
if err != nil {
slog.Error("handleEC2RunInstances vmMgr.Run failed", "instanceId", instance.ID, "err", err)
if instance.GPUPCIAddress != "" && d.gpuManager != nil {
if releaseErr := d.gpuManager.Release(instance.ID); releaseErr != nil {
slog.Error("handleEC2RunInstances: GPU release failed after launch failure",
"instanceId", instance.ID, "err", releaseErr)
}
}
d.vmMgr.MarkFailed(instance, "launch_failed")
continue
}
// Discover actual guest device names via QMP query-block
d.vmMgr.UpdateGuestDeviceNames(instance)
successCount++
slog.Info("handleEC2RunInstances launched instance", "instanceId", instance.ID)
}
slog.Info("handleEC2RunInstances completed", "requested", launchCount, "created", len(instances), "launched", successCount)
}
func (d *Daemon) handleRebootInstance(msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) {
slog.Info("Rebooting instance", "id", command.ID)
if err := d.vmMgr.Reboot(instance.ID); err != nil {
switch {
case errors.Is(err, vm.ErrInstanceNotFound):
respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
case errors.Is(err, vm.ErrInvalidTransition):
slog.Error("RebootInstance: instance not in running state",
"instanceId", command.ID, "err", err)
respondWithError(msg, awserrors.ErrorIncorrectInstanceState)
default:
slog.Error("RebootInstance: reboot failed", "instanceId", command.ID, "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
}
return
}
slog.Info("Instance rebooted", "instanceId", command.ID)
if err := msg.Respond([]byte(`{}`)); err != nil {
slog.Error("Failed to respond to NATS request", "err", err)
}
}
func (d *Daemon) handleStartInstance(msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) {
slog.Info("Starting instance", "id", command.ID)
// Validate instance is in stopped state
status := d.vmMgr.Status(instance)
if status != vm.StateStopped {
slog.Error("StartInstance: instance not in stopped state", "instanceId", command.ID, "status", status)
respondWithError(msg, awserrors.ErrorIncorrectInstanceState)
return
}
// Allocate resources
instanceType, ok := d.resourceMgr.instanceTypes[instance.InstanceType]
if ok {
if err := d.resourceMgr.allocate(instanceType); err != nil {
slog.Error("Failed to allocate resources for start command", "id", command.ID, "err", err)
respondWithError(msg, awserrors.ErrorInsufficientInstanceCapacity)
return
}
}
// Clear stop attribute before launch so WriteState inside the manager
// persists the correct attributes. Without this, a daemon restart after
// a stop→start cycle would see StopInstance=true and skip reconnecting QEMU.
d.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Attributes = command.Attributes })
// Launch the instance infrastructure (QEMU, QMP, NATS subscriptions)
if err := d.vmMgr.Start(instance.ID); err != nil {
slog.Error("handleStartInstance: vmMgr.Start failed", "err", err)
if ok {
d.resourceMgr.deallocate(instanceType)
}
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
// Discover actual guest device names via QMP query-block
d.vmMgr.UpdateGuestDeviceNames(instance)
slog.Info("Instance started", "instanceId", instance.ID)
if err := msg.Respond(fmt.Appendf(nil, `{"status":"running","instanceId":"%s"}`, instance.ID)); err != nil {
slog.Error("Failed to respond to NATS request", "err", err)
}
}
func (d *Daemon) handleStopOrTerminateInstance(msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) {
isTerminate := command.Attributes.TerminateInstance
action := "Stopping"
initialState := vm.StateStopping
if isTerminate {
action = "Terminating"
initialState = vm.StateShuttingDown
}
slog.Info(action+" instance", "id", command.ID)
currentState := d.vmMgr.Status(instance)
// Idempotent: a concurrent terminate goroutine is already cleaning up.
if isTerminate && currentState == vm.StateShuttingDown {
slog.Info("Instance already shutting down, terminate is idempotent", "instanceId", instance.ID)
if err := msg.Respond([]byte(`{}`)); err != nil {
slog.Error("Failed to respond to NATS request", "err", err)
}
return
}
// Validate the transition synchronously before dispatching so the AWS
// SDK sees IncorrectInstanceState (400) instead of a stale 200. The
// async Stop/Terminate path re-validates and surfaces vm.ErrInvalidTransition
// on a racing transition; we map that into the same AWS error below.
if !vm.IsValidTransition(currentState, initialState) {
slog.Warn("Instance in incorrect state for "+strings.ToLower(action),
"instanceId", instance.ID, "currentState", string(currentState))
respondWithError(msg, awserrors.ErrorIncorrectInstanceState)
return
}
// Stamp the command attributes onto the VM before dispatch so the persisted
// state reflects the user-stop / user-terminate intent (e.g. for the
// recovery path that distinguishes user-stopped from crash-stopped).
d.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Attributes = command.Attributes })
// Respond immediately - cleanup completes in background.
if err := msg.Respond([]byte(`{}`)); err != nil {
slog.Error("Failed to respond to NATS request", "err", err)
}
go func(id string) {
var err error
if isTerminate {
err = d.vmMgr.Terminate(id)
} else {
err = d.vmMgr.Stop(id)
}
if err != nil {
// Race with another transition: log at the right level so it
// shows up in dashboards but doesn't trigger paging.
if errors.Is(err, vm.ErrInvalidTransition) {
slog.Warn("Lifecycle transition raced; ack already sent",
"id", id, "action", strings.ToLower(action), "err", err)
return
}
slog.Error("Failed to "+strings.ToLower(action)+" instance", "err", err, "id", id)
}
}(instance.ID)
d.instanceService.LaunchRunInstances(instances, input, instanceType)
}
// describeInstancesValidFilters defines the set of filter names accepted by DescribeInstances.
@@ -779,277 +314,12 @@ func (d *Daemon) handleEC2DescribeInstanceTypes(msg *nats.Msg) {
}
// startStoppedInstanceRequest is the payload for ec2.start topic
type startStoppedInstanceRequest struct {
InstanceID string `json:"instance_id"`
}
// handleEC2StartStoppedInstance picks up a stopped instance from shared KV,
// re-launches it on this daemon node, and removes it from shared KV.
func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) {
var req startStoppedInstanceRequest
if err := json.Unmarshal(msg.Data, &req); err != nil {
slog.Error("handleEC2StartStoppedInstance: failed to unmarshal request", "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
if req.InstanceID == "" {
slog.Error("handleEC2StartStoppedInstance: missing instance_id")
respondWithError(msg, awserrors.ErrorMissingParameter)
return
}
if d.stateStore == nil {
slog.Error("handleEC2StartStoppedInstance: state store not available")
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
// Load instance from shared KV
instance, err := d.stateStore.LoadStoppedInstance(req.InstanceID)
if err != nil {
slog.Error("handleEC2StartStoppedInstance: failed to load stopped instance", "instanceId", req.InstanceID, "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
if instance == nil {
slog.Warn("handleEC2StartStoppedInstance: instance not found in shared KV", "instanceId", req.InstanceID)
respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
return
}
if instance.Status != vm.StateStopped {
slog.Error("handleEC2StartStoppedInstance: instance not in stopped state", "instanceId", req.InstanceID, "status", instance.Status)
respondWithError(msg, awserrors.ErrorIncorrectInstanceState)
return
}
// Verify the caller owns this instance
if !checkInstanceOwnership(msg, req.InstanceID, instance.AccountID) {
return
}
// Reset node-local fields that are stale after cross-node migration
instance.ResetNodeLocalState()
// Allocate resources
instanceType, ok := d.resourceMgr.instanceTypes[instance.InstanceType]
if !ok {
slog.Error("handleEC2StartStoppedInstance: instance type not available on this node",
"instanceId", req.InstanceID, "instanceType", instance.InstanceType)
respondWithError(msg, awserrors.ErrorInsufficientInstanceCapacity)
return
}
if err := d.resourceMgr.allocate(instanceType); err != nil {
slog.Error("handleEC2StartStoppedInstance: failed to allocate resources", "instanceId", req.InstanceID, "err", err)
respondWithError(msg, awserrors.ErrorInsufficientInstanceCapacity)
return
}
// Add instance to local map and clear stop attribute before launch
instance.Attributes = types.EC2CommandAttributes{StartInstance: true}
d.vmMgr.Insert(instance)
// Claim GPU for GPU instance types — binds the full IOMMU group to vfio-pci.
if d.gpuManager != nil && instancetypes.IsGPUType(instanceType) {
gpuDevice, gpuErr := d.gpuManager.Claim(instance.ID)
if gpuErr != nil {
slog.Error("handleEC2StartStoppedInstance: GPU claim failed", "instanceId", req.InstanceID, "err", gpuErr)
d.resourceMgr.deallocate(instanceType)
d.vmMgr.Delete(instance.ID)
respondWithError(msg, awserrors.ErrorInsufficientInstanceCapacity)
return
}
instance.GPUPCIAddress = gpuDevice.PCIAddress
instance.GPUXVGAEnabled = gpuXVGAEnabled(gpuDevice, d.config.Daemon.GPUModelOverrides)
slog.Info("GPU claimed for instance", "instanceId", req.InstanceID, "gpu", gpuDevice.PCIAddress, "xvga", instance.GPUXVGAEnabled)
}
// Launch the instance infrastructure (QEMU, QMP, NATS subscriptions)
err = d.vmMgr.Run(instance)
if err != nil {
slog.Error("handleEC2StartStoppedInstance: vmMgr.Run failed", "instanceId", req.InstanceID, "err", err)
// Rollback: release GPU if claimed, deallocate resources, remove from local map
if instance.GPUPCIAddress != "" && d.gpuManager != nil {
if releaseErr := d.gpuManager.Release(instance.ID); releaseErr != nil {
slog.Error("handleEC2StartStoppedInstance: GPU release failed after launch failure",
"instanceId", req.InstanceID, "err", releaseErr)
}
}
if ok {
d.resourceMgr.deallocate(instanceType)
}
d.vmMgr.Delete(instance.ID)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
// Discover actual guest device names via QMP query-block
d.vmMgr.UpdateGuestDeviceNames(instance)
// Remove from shared KV now that it's running locally.
// Retry once on failure — a stale KV entry risks duplicate starts.
if err := d.stateStore.DeleteStoppedInstance(req.InstanceID); err != nil {
slog.Warn("handleEC2StartStoppedInstance: first KV delete failed, retrying",
"instanceId", req.InstanceID, "err", err)
if retryErr := d.stateStore.DeleteStoppedInstance(req.InstanceID); retryErr != nil {
slog.Error("handleEC2StartStoppedInstance: KV delete failed after retry, instance is running locally but stale entry remains in shared KV",
"instanceId", req.InstanceID, "err", retryErr)
}
}
slog.Info("Started stopped instance from shared KV", "instanceId", instance.ID, "node", d.node)
if err := msg.Respond(fmt.Appendf(nil, `{"status":"running","instanceId":"%s"}`, instance.ID)); err != nil {
slog.Error("Failed to respond to NATS request", "err", err)
}
handleNATSRequest(msg, d.instanceService.StartStoppedInstance)
}
// terminateStoppedInstanceRequest is the payload for ec2.terminate topic
type terminateStoppedInstanceRequest struct {
InstanceID string `json:"instance_id"`
}
// handleEC2TerminateStoppedInstance picks up a stopped instance from shared KV,
// deletes its volumes, and removes it from shared KV.
func (d *Daemon) handleEC2TerminateStoppedInstance(msg *nats.Msg) {
var req terminateStoppedInstanceRequest
if err := json.Unmarshal(msg.Data, &req); err != nil {
slog.Error("handleEC2TerminateStoppedInstance: failed to unmarshal request", "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
if req.InstanceID == "" {
slog.Error("handleEC2TerminateStoppedInstance: missing instance_id")
respondWithError(msg, awserrors.ErrorMissingParameter)
return
}
if d.stateStore == nil {
slog.Error("handleEC2TerminateStoppedInstance: state store not available")
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
// Load instance from shared KV
instance, err := d.stateStore.LoadStoppedInstance(req.InstanceID)
if err != nil {
slog.Error("handleEC2TerminateStoppedInstance: failed to load stopped instance", "instanceId", req.InstanceID, "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
if instance == nil {
slog.Warn("handleEC2TerminateStoppedInstance: instance not found in shared KV", "instanceId", req.InstanceID)
respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
return
}
if instance.Status != vm.StateStopped {
slog.Error("handleEC2TerminateStoppedInstance: instance not in stopped state", "instanceId", req.InstanceID, "status", instance.Status)
respondWithError(msg, awserrors.ErrorIncorrectInstanceState)
return
}
// Verify the caller owns this instance
if !checkInstanceOwnership(msg, req.InstanceID, instance.AccountID) {
return
}
// Delete volumes — no QEMU shutdown or unmount needed (already done during stop)
instance.EBSRequests.Mu.Lock()
for _, ebsRequest := range instance.EBSRequests.Requests {
// Internal volumes (EFI, cloud-init) are always cleaned up via ebs.delete
if ebsRequest.EFI || ebsRequest.CloudInit {
ebsDeleteData, err := json.Marshal(types.EBSDeleteRequest{Volume: ebsRequest.Name})
if err != nil {
slog.Error("handleEC2TerminateStoppedInstance: failed to marshal ebs.delete request", "name", ebsRequest.Name, "err", err)
continue
}
deleteMsg, err := d.natsConn.Request("ebs.delete", ebsDeleteData, 30*time.Second)
if err != nil {
slog.Warn("handleEC2TerminateStoppedInstance: ebs.delete failed for internal volume", "name", ebsRequest.Name, "err", err)
} else {
slog.Info("handleEC2TerminateStoppedInstance: ebs.delete sent for internal volume", "name", ebsRequest.Name, "data", string(deleteMsg.Data))
}
continue
}
// User-visible volumes: respect DeleteOnTermination flag
if !ebsRequest.DeleteOnTermination {
slog.Info("handleEC2TerminateStoppedInstance: volume has DeleteOnTermination=false, skipping", "name", ebsRequest.Name)
continue
}
slog.Info("handleEC2TerminateStoppedInstance: deleting volume with DeleteOnTermination=true", "name", ebsRequest.Name)
_, err := d.volumeService.DeleteVolume(&ec2.DeleteVolumeInput{
VolumeId: &ebsRequest.Name,
}, instance.AccountID)
if err != nil {
slog.Error("handleEC2TerminateStoppedInstance: failed to delete volume", "name", ebsRequest.Name, "err", err)
}
}
instance.EBSRequests.Mu.Unlock()
// Release public IP before termination
if instance.PublicIP != "" && instance.PublicIPPool != "" && d.externalIPAM != nil {
portName := "port-" + instance.ENIId
vpcId := ""
logicalIP := ""
if instance.Instance != nil {
if instance.Instance.VpcId != nil {
vpcId = *instance.Instance.VpcId
}
if instance.Instance.PrivateIpAddress != nil {
logicalIP = *instance.Instance.PrivateIpAddress
}
}
d.publishNATEvent("vpc.delete-nat", vpcId, instance.PublicIP, logicalIP, portName, "")
if err := d.externalIPAM.ReleaseIP(instance.PublicIPPool, instance.PublicIP); err != nil {
slog.Warn("handleEC2TerminateStoppedInstance: failed to release public IP", "ip", instance.PublicIP, "pool", instance.PublicIPPool, "err", err)
} else {
slog.Info("handleEC2TerminateStoppedInstance: released public IP", "ip", instance.PublicIP, "instanceId", req.InstanceID)
}
}
// Delete ENI if present
if instance.ENIId != "" && d.vpcService != nil {
_, eniErr := d.vpcService.DeleteNetworkInterface(&ec2.DeleteNetworkInterfaceInput{
NetworkInterfaceId: &instance.ENIId,
}, instance.AccountID)
if eniErr != nil {
slog.Error("handleEC2TerminateStoppedInstance: failed to delete ENI", "eni", instance.ENIId, "err", eniErr)
} else {
slog.Info("handleEC2TerminateStoppedInstance: deleted ENI", "eni", instance.ENIId, "instanceId", req.InstanceID)
}
}
// Write to terminated KV bucket FIRST so the instance is visible in DescribeInstances.
// If this fails, the instance remains in the stopped bucket (safe to retry).
instance.Status = vm.StateTerminated
if err := d.stateStore.WriteTerminatedInstance(req.InstanceID, instance); err != nil {
slog.Error("handleEC2TerminateStoppedInstance: failed to write to terminated KV, aborting", "instanceId", req.InstanceID, "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
// Now safe to remove from shared stopped KV — instance already exists in terminated bucket.
// Retry once on failure to avoid duplicate entries in DescribeInstances.
if err := d.stateStore.DeleteStoppedInstance(req.InstanceID); err != nil {
slog.Warn("handleEC2TerminateStoppedInstance: first stopped KV delete failed, retrying",
"instanceId", req.InstanceID, "err", err)
if retryErr := d.stateStore.DeleteStoppedInstance(req.InstanceID); retryErr != nil {
slog.Error("handleEC2TerminateStoppedInstance: stopped KV delete failed after retry, instance may appear in both buckets",
"instanceId", req.InstanceID, "err", retryErr)
}
}
slog.Info("Terminated stopped instance from shared KV", "instanceId", req.InstanceID)
if err := msg.Respond(fmt.Appendf(nil, `{"status":"terminated","instanceId":"%s"}`, req.InstanceID)); err != nil {
slog.Error("Failed to respond to NATS request", "err", err)
}
handleNATSRequest(msg, d.instanceService.TerminateStoppedInstance)
}
// handleEC2DescribeStoppedInstances returns stopped instances from shared KV.
@@ -1162,110 +432,8 @@ func (d *Daemon) describeInstancesFromKV(msg *nats.Msg, listFn func() ([]*vm.VM,
slog.Info(handlerName+" completed", "count", len(reservations))
}
// handleEC2ModifyInstanceAttribute modifies attributes of a stopped instance in shared KV.
// All supported attributes (InstanceType, UserData) require the instance to be stopped.
func (d *Daemon) handleEC2ModifyInstanceAttribute(msg *nats.Msg) {
var input ec2.ModifyInstanceAttributeInput
if err := json.Unmarshal(msg.Data, &input); err != nil {
slog.Error("handleEC2ModifyInstanceAttribute: failed to unmarshal request", "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
if input.InstanceId == nil || *input.InstanceId == "" {
slog.Error("handleEC2ModifyInstanceAttribute: missing instance_id")
respondWithError(msg, awserrors.ErrorMissingParameter)
return
}
instanceID := *input.InstanceId
// SourceDestCheck is a networking concept that doesn't apply to bare-metal VMs.
// Accept the call as a no-op so Terraform and the AWS CLI don't error out.
// Unlike InstanceType/UserData, AWS allows this on running instances, so handle
// it before the stopped-state gate.
if input.SourceDestCheck != nil {
slog.Info("handleEC2ModifyInstanceAttribute: accepting SourceDestCheck (no-op on bare metal)", "instanceId", instanceID)
if err := msg.Respond([]byte(`{}`)); err != nil {
slog.Error("Failed to respond to NATS request", "err", err)
}
return
}
if d.stateStore == nil {
slog.Error("handleEC2ModifyInstanceAttribute: state store not available")
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
instance, err := d.stateStore.LoadStoppedInstance(instanceID)
if err != nil {
slog.Error("handleEC2ModifyInstanceAttribute: failed to load stopped instance", "instanceId", instanceID, "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
if instance == nil {
slog.Warn("handleEC2ModifyInstanceAttribute: instance not found in shared KV", "instanceId", instanceID)
respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
return
}
if instance.Status != vm.StateStopped {
slog.Error("handleEC2ModifyInstanceAttribute: instance not in stopped state", "instanceId", instanceID, "status", instance.Status)
respondWithError(msg, awserrors.ErrorIncorrectInstanceState)
return
}
// Verify the caller owns this instance
if !checkInstanceOwnership(msg, instanceID, instance.AccountID) {
return
}
// Apply the requested attribute change
if input.InstanceType != nil && input.InstanceType.Value != nil {
newType := *input.InstanceType.Value
if newType == "" {
slog.Error("handleEC2ModifyInstanceAttribute: empty instance type value", "instanceId", instanceID)
respondWithError(msg, awserrors.ErrorInvalidInstanceAttributeValue)
return
}
if instance.Instance == nil {
slog.Error("handleEC2ModifyInstanceAttribute: instance.Instance is nil, data integrity issue", "instanceId", instanceID)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
slog.Info("handleEC2ModifyInstanceAttribute: changing instance type",
"instanceId", instanceID, "oldType", instance.InstanceType, "newType", newType)
instance.InstanceType = newType
instance.Config.InstanceType = newType
instance.Instance.InstanceType = aws.String(newType)
// Clear StateReason — resolves capacity-unavailable state from instance-type-missing bug
instance.Instance.StateReason = nil
}
if input.UserData != nil && input.UserData.Value != nil {
slog.Info("handleEC2ModifyInstanceAttribute: changing user data", "instanceId", instanceID)
// Value arrives as decoded bytes (JSON unmarshal handles base64 → []byte automatically)
instance.UserData = string(input.UserData.Value)
if instance.RunInstancesInput != nil {
instance.RunInstancesInput.UserData = aws.String(base64.StdEncoding.EncodeToString(input.UserData.Value))
}
}
if err := d.stateStore.WriteStoppedInstance(instanceID, instance); err != nil {
slog.Error("handleEC2ModifyInstanceAttribute: failed to write modified instance to KV",
"instanceId", instanceID, "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
slog.Info("handleEC2ModifyInstanceAttribute: completed successfully", "instanceId", instanceID)
if err := msg.Respond([]byte(`{}`)); err != nil {
slog.Error("Failed to respond to NATS request", "err", err)
}
handleNATSRequest(msg, d.instanceService.ModifyInstanceAttribute)
}
// handleEC2DescribeInstanceAttribute returns a single requested attribute for an instance.
@@ -1380,26 +548,3 @@ func (d *Daemon) handleEC2DescribeInstanceAttribute(msg *nats.Msg) {
"instanceId", instanceID, "attribute", attribute, "accountID", accountID)
respondWithJSON(msg, output)
}
// publishNATEvent sends a NAT lifecycle event (vpc.add-nat or vpc.delete-nat) to NATS.
// For vpc.add-nat, it uses request-reply to ensure the OVN NAT rule is committed
// before returning, preventing ARP propagation races. For vpc.delete-nat, it
// uses fire-and-forget since the caller doesn't need to wait.
func (d *Daemon) publishNATEvent(topic, vpcId, externalIP, logicalIP, portName, mac string) {
evt := struct {
VpcId string `json:"vpc_id"`
ExternalIP string `json:"external_ip"`
LogicalIP string `json:"logical_ip"`
PortName string `json:"port_name"`
MAC string `json:"mac"`
}{VpcId: vpcId, ExternalIP: externalIP, LogicalIP: logicalIP, PortName: portName, MAC: mac}
if topic == "vpc.add-nat" {
if err := utils.RequestEvent(d.natsConn, topic, evt, 10*time.Second); err != nil {
slog.Warn("publishNATEvent: failed to add NAT rule — OVN dnat_and_snat rule not created; restart vpcd or re-associate EIP to recover",
"topic", topic, "externalIP", externalIP, "logicalIP", logicalIP, "err", err)
}
return
}
utils.PublishEvent(d.natsConn, topic, evt)
}
spinifex/daemon/daemon_handlers_instance_test.go
+19 -11
View File
@@ -10,6 +10,8 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/awserrors"
handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance"
"github.com/mulgadc/spinifex/spinifex/objectstore"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/vm"
"github.com/nats-io/nats.go"
@@ -142,11 +144,17 @@ var _ vm.StateStore = (*fakeStateStore)(nil)
// daemonWithFakeStateStore returns a daemon wired with an in-memory NATS
// connection (via createTestDaemon) and the supplied fake StateStore.
// The daemon does not have JetStream initialized.
// The daemon does not have JetStream initialized. Rewires d.instanceService
// to point at the fake store so handlers that delegate to InstanceService
// (e.g. ModifyInstanceAttribute) see the injected state.
func daemonWithFakeStateStore(t *testing.T, store *fakeStateStore) *Daemon {
t.Helper()
d := createTestDaemon(t, sharedNATSURL)
d.stateStore = store
d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(
d.config, d.resourceMgr.instanceTypes, d.natsConn,
objectstore.NewMemoryObjectStore(), d.vmMgr, d.resourceMgr, store,
)
return d
}
@@ -199,7 +207,7 @@ func TestHandleEC2StartStoppedInstance_LoadError(t *testing.T) {
store.loadStoppedErr = errors.New("kv unavailable")
d := daemonWithFakeStateStore(t, store)
body, _ := json.Marshal(startStoppedInstanceRequest{InstanceID: "i-load-fail"})
body, _ := json.Marshal(handlers_ec2_instance.StartStoppedInstanceInput{InstanceID: "i-load-fail"})
reply := requestHandler(t, d.natsConn, "ec2.start.test1", d.handleEC2StartStoppedInstance, testAccountID, body)
assert.Equal(t, awserrors.ErrorServerInternal, decodeError(t, reply.Data)["Code"])
}
@@ -208,7 +216,7 @@ func TestHandleEC2StartStoppedInstance_StateStoreNil(t *testing.T) {
d := createTestDaemon(t, sharedNATSURL)
// d.stateStore intentionally left nil.
body, _ := json.Marshal(startStoppedInstanceRequest{InstanceID: "i-no-store"})
body, _ := json.Marshal(handlers_ec2_instance.StartStoppedInstanceInput{InstanceID: "i-no-store"})
reply := requestHandler(t, d.natsConn, "ec2.start.test2", d.handleEC2StartStoppedInstance, testAccountID, body)
assert.Equal(t, awserrors.ErrorServerInternal, decodeError(t, reply.Data)["Code"])
}
@@ -218,7 +226,7 @@ func TestHandleEC2StartStoppedInstance_CrossTenantRejected(t *testing.T) {
store.stopped["i-foreign"] = stoppedVMFixture("i-foreign", "999988887777")
d := daemonWithFakeStateStore(t, store)
body, _ := json.Marshal(startStoppedInstanceRequest{InstanceID: "i-foreign"})
body, _ := json.Marshal(handlers_ec2_instance.StartStoppedInstanceInput{InstanceID: "i-foreign"})
reply := requestHandler(t, d.natsConn, "ec2.start.test3", d.handleEC2StartStoppedInstance, testAccountID, body)
assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, decodeError(t, reply.Data)["Code"])
@@ -237,7 +245,7 @@ func TestHandleEC2StartStoppedInstance_InstanceTypeUnknown(t *testing.T) {
store.stopped[v.ID] = v
d := daemonWithFakeStateStore(t, store)
body, _ := json.Marshal(startStoppedInstanceRequest{InstanceID: v.ID})
body, _ := json.Marshal(handlers_ec2_instance.StartStoppedInstanceInput{InstanceID: v.ID})
reply := requestHandler(t, d.natsConn, "ec2.start.test4", d.handleEC2StartStoppedInstance, testAccountID, body)
assert.Equal(t, awserrors.ErrorInsufficientInstanceCapacity, decodeError(t, reply.Data)["Code"])
}
@@ -249,7 +257,7 @@ func TestHandleEC2TerminateStoppedInstance_LoadError(t *testing.T) {
store.loadStoppedErr = errors.New("kv unavailable")
d := daemonWithFakeStateStore(t, store)
body, _ := json.Marshal(terminateStoppedInstanceRequest{InstanceID: "i-load-fail"})
body, _ := json.Marshal(handlers_ec2_instance.TerminateStoppedInstanceInput{InstanceID: "i-load-fail"})
reply := requestHandler(t, d.natsConn, "ec2.terminate.test1", d.handleEC2TerminateStoppedInstance, testAccountID, body)
assert.Equal(t, awserrors.ErrorServerInternal, decodeError(t, reply.Data)["Code"])
}
@@ -257,7 +265,7 @@ func TestHandleEC2TerminateStoppedInstance_LoadError(t *testing.T) {
func TestHandleEC2TerminateStoppedInstance_StateStoreNil(t *testing.T) {
d := createTestDaemon(t, sharedNATSURL)
body, _ := json.Marshal(terminateStoppedInstanceRequest{InstanceID: "i-no-store"})
body, _ := json.Marshal(handlers_ec2_instance.TerminateStoppedInstanceInput{InstanceID: "i-no-store"})
reply := requestHandler(t, d.natsConn, "ec2.terminate.test2", d.handleEC2TerminateStoppedInstance, testAccountID, body)
assert.Equal(t, awserrors.ErrorServerInternal, decodeError(t, reply.Data)["Code"])
}
@@ -271,7 +279,7 @@ func TestHandleEC2TerminateStoppedInstance_WriteTerminatedFailureAborts(t *testi
store.stopped[v.ID] = v
d := daemonWithFakeStateStore(t, store)
body, _ := json.Marshal(terminateStoppedInstanceRequest{InstanceID: v.ID})
body, _ := json.Marshal(handlers_ec2_instance.TerminateStoppedInstanceInput{InstanceID: v.ID})
reply := requestHandler(t, d.natsConn, "ec2.terminate.test3", d.handleEC2TerminateStoppedInstance, testAccountID, body)
assert.Equal(t, awserrors.ErrorServerInternal, decodeError(t, reply.Data)["Code"])
@@ -294,7 +302,7 @@ func TestHandleEC2TerminateStoppedInstance_DeleteRetrySucceeds(t *testing.T) {
store.stopped[v.ID] = v
d := daemonWithFakeStateStore(t, store)
body, _ := json.Marshal(terminateStoppedInstanceRequest{InstanceID: v.ID})
body, _ := json.Marshal(handlers_ec2_instance.TerminateStoppedInstanceInput{InstanceID: v.ID})
reply := requestHandler(t, d.natsConn, "ec2.terminate.test4", d.handleEC2TerminateStoppedInstance, testAccountID, body)
var resp map[string]string
@@ -321,7 +329,7 @@ func TestHandleEC2TerminateStoppedInstance_DeleteAlwaysFailsKeepsTerminated(t *t
store.stopped[v.ID] = v
d := daemonWithFakeStateStore(t, store)
body, _ := json.Marshal(terminateStoppedInstanceRequest{InstanceID: v.ID})
body, _ := json.Marshal(handlers_ec2_instance.TerminateStoppedInstanceInput{InstanceID: v.ID})
reply := requestHandler(t, d.natsConn, "ec2.terminate.test5", d.handleEC2TerminateStoppedInstance, testAccountID, body)
var resp map[string]string
@@ -339,7 +347,7 @@ func TestHandleEC2TerminateStoppedInstance_CrossTenantRejected(t *testing.T) {
store.stopped["i-foreign-term"] = stoppedVMFixture("i-foreign-term", "999988887777")
d := daemonWithFakeStateStore(t, store)
body, _ := json.Marshal(terminateStoppedInstanceRequest{InstanceID: "i-foreign-term"})
body, _ := json.Marshal(handlers_ec2_instance.TerminateStoppedInstanceInput{InstanceID: "i-foreign-term"})
reply := requestHandler(t, d.natsConn, "ec2.terminate.test6", d.handleEC2TerminateStoppedInstance, testAccountID, body)
assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, decodeError(t, reply.Data)["Code"])
spinifex/daemon/daemon_handlers_test.go
+22 -430
View File
@@ -65,6 +65,11 @@ func createFullTestDaemonWithStore(t *testing.T, natsURL string) (*Daemon, *obje
daemon.tagsService = handlers_ec2_tags.NewTagsServiceImplWithStore(cfg, memStore)
initAccountServiceForTest(t, daemon)
// Wire RunInstances deps now that image/key services exist. vpcService and
// externalIPAM are nil in tests, so PrepareRunInstances will skip the
// subnet/ENI/public-IP code paths.
daemon.instanceService.SetRunInstancesDeps(daemon.imageService, daemon.keyService, nil, nil)
return daemon, memStore
}
@@ -96,6 +101,7 @@ func createFullTestDaemonWithJetStream(t *testing.T, natsURL string) *Daemon {
objectstore.NewMemoryObjectStore(),
daemon.vmMgr, daemon.resourceMgr, daemon.jsManager,
)
daemon.instanceService.SetRunInstancesDeps(daemon.imageService, daemon.keyService, nil, nil)
return daemon
}
@@ -1363,121 +1369,13 @@ func TestHandleEC2DescribeStoppedInstances_WithFilter(t *testing.T) {
_ = daemon.jsManager.DeleteStoppedInstance("i-filter-002")
}
// --- handleEC2TerminateStoppedInstance tests ---
func TestHandleEC2TerminateStoppedInstance_MissingInstanceID(t *testing.T) {
natsURL := sharedJSNATSURL
daemon := createFullTestDaemonWithJetStream(t, natsURL)
sub, err := daemon.natsConn.QueueSubscribe("ec2.terminate", "spinifex-workers", daemon.handleEC2TerminateStoppedInstance)
require.NoError(t, err)
defer sub.Unsubscribe()
reqData, _ := json.Marshal(map[string]string{"instance_id": ""})
reply, err := daemon.natsConn.Request("ec2.terminate", reqData, 5*time.Second)
require.NoError(t, err)
var errResp map[string]any
err = json.Unmarshal(reply.Data, &errResp)
require.NoError(t, err)
assert.Equal(t, awserrors.ErrorMissingParameter, errResp["Code"])
}
func TestHandleEC2TerminateStoppedInstance_MissingInstance(t *testing.T) {
natsURL := sharedJSNATSURL
daemon := createFullTestDaemonWithJetStream(t, natsURL)
sub, err := daemon.natsConn.QueueSubscribe("ec2.terminate", "spinifex-workers", daemon.handleEC2TerminateStoppedInstance)
require.NoError(t, err)
defer sub.Unsubscribe()
reqData, _ := json.Marshal(map[string]string{"instance_id": "i-nonexistent"})
reply, err := daemon.natsConn.Request("ec2.terminate", reqData, 5*time.Second)
require.NoError(t, err)
var errResp map[string]any
err = json.Unmarshal(reply.Data, &errResp)
require.NoError(t, err)
assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, errResp["Code"])
}
func TestHandleEC2TerminateStoppedInstance_NotStoppedState(t *testing.T) {
natsURL := sharedJSNATSURL
daemon := createFullTestDaemonWithJetStream(t, natsURL)
// Write an instance in running state to shared KV
runningVM := &vm.VM{
ID: "i-term-running",
Status: vm.StateRunning,
InstanceType: getTestInstanceType(t),
AccountID: testAccountID,
}
err := daemon.jsManager.WriteStoppedInstance(runningVM.ID, runningVM)
require.NoError(t, err)
sub, err := daemon.natsConn.QueueSubscribe("ec2.terminate", "spinifex-workers", daemon.handleEC2TerminateStoppedInstance)
require.NoError(t, err)
defer sub.Unsubscribe()
reqData, _ := json.Marshal(map[string]string{"instance_id": "i-term-running"})
reply, err := natsRequest(daemon.natsConn, "ec2.terminate", reqData, 5*time.Second)
require.NoError(t, err)
var errResp map[string]any
err = json.Unmarshal(reply.Data, &errResp)
require.NoError(t, err)
assert.Equal(t, awserrors.ErrorIncorrectInstanceState, errResp["Code"])
// Cleanup
_ = daemon.jsManager.DeleteStoppedInstance(runningVM.ID)
}
func TestHandleEC2TerminateStoppedInstance_Success(t *testing.T) {
natsURL := sharedJSNATSURL
daemon := createFullTestDaemonWithJetStream(t, natsURL)
// Write a stopped instance to shared KV
stoppedVM := &vm.VM{
ID: "i-term-stopped-001",
Status: vm.StateStopped,
InstanceType: getTestInstanceType(t),
LastNode: "node-1",
AccountID: testAccountID,
Reservation: &ec2.Reservation{
ReservationId: aws.String("r-term-001"),
OwnerId: aws.String("123456789012"),
},
Instance: &ec2.Instance{
InstanceId: aws.String("i-term-stopped-001"),
InstanceType: aws.String(getTestInstanceType(t)),
},
}
err := daemon.jsManager.WriteStoppedInstance(stoppedVM.ID, stoppedVM)
require.NoError(t, err)
sub, err := daemon.natsConn.QueueSubscribe("ec2.terminate", "spinifex-workers", daemon.handleEC2TerminateStoppedInstance)
require.NoError(t, err)
defer sub.Unsubscribe()
reqData, _ := json.Marshal(map[string]string{"instance_id": "i-term-stopped-001"})
reply, err := natsRequest(daemon.natsConn, "ec2.terminate", reqData, 5*time.Second)
require.NoError(t, err)
var resp map[string]string
err = json.Unmarshal(reply.Data, &resp)
require.NoError(t, err)
assert.Equal(t, "terminated", resp["status"])
assert.Equal(t, "i-term-stopped-001", resp["instanceId"])
// Verify instance was removed from shared KV
loaded, err := daemon.jsManager.LoadStoppedInstance("i-term-stopped-001")
require.NoError(t, err)
assert.Nil(t, loaded, "Instance should be removed from shared KV after termination")
}
// --- handleEC2TerminateStoppedInstance wrapper smoke test ---
//
// Detailed logic coverage lives in
// handlers/ec2/instance/service_impl_test.go (TestTerminateStoppedInstance_*).
// The wrapper smoke case is TestHandleEC2TerminateStoppedInstance_WritesToTerminatedKV
// further below; it confirms the NATS → handleNATSRequest → service round-trip
// stays intact end-to-end against real JetStream KV.
func TestHandleEC2GetConsoleOutput(t *testing.T) {
natsURL := sharedNATSURL
@@ -1653,9 +1551,14 @@ func TestAttachVolume_ZoneMismatch(t *testing.T) {
assert.Contains(t, string(resp.Data), "InvalidVolume.ZoneMismatch")
}
// --- handleEC2ModifyInstanceAttribute tests ---
// --- handleEC2ModifyInstanceAttribute wrapper smoke test ---
//
// Detailed logic coverage lives in
// handlers/ec2/instance/service_impl_test.go (TestModifyInstanceAttribute_*).
// This case keeps one end-to-end NATS → handleNATSRequest → service round-trip
// to confirm the daemon wiring stays intact.
func TestHandleEC2ModifyInstanceAttribute_ChangeInstanceType(t *testing.T) {
func TestHandleEC2ModifyInstanceAttribute_WrapperRoundTrip(t *testing.T) {
natsURL := sharedJSNATSURL
daemon := createFullTestDaemonWithJetStream(t, natsURL)
@@ -1664,7 +1567,7 @@ func TestHandleEC2ModifyInstanceAttribute_ChangeInstanceType(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
instanceID := "i-modify-type-001"
instanceID := "i-modify-wrapper-001"
instance := &vm.VM{
ID: instanceID,
Status: vm.StateStopped,
@@ -1693,260 +1596,6 @@ func TestHandleEC2ModifyInstanceAttribute_ChangeInstanceType(t *testing.T) {
require.NoError(t, err)
require.NotNil(t, updated)
assert.Equal(t, "t3.medium", updated.InstanceType)
assert.Equal(t, "t3.medium", updated.Config.InstanceType)
assert.Equal(t, "t3.medium", *updated.Instance.InstanceType)
}
func TestHandleEC2ModifyInstanceAttribute_ChangeUserData(t *testing.T) {
natsURL := sharedJSNATSURL
daemon := createFullTestDaemonWithJetStream(t, natsURL)
sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", daemon.handleEC2ModifyInstanceAttribute)
require.NoError(t, err)
defer sub.Unsubscribe()
instanceID := "i-modify-ud-001"
instance := &vm.VM{
ID: instanceID,
Status: vm.StateStopped,
InstanceType: "t3.micro",
AccountID: testAccountID,
UserData: "old data",
RunInstancesInput: &ec2.RunInstancesInput{
UserData: aws.String("b2xkIGRhdGE="),
},
Instance: &ec2.Instance{
InstanceId: aws.String(instanceID),
},
}
err = daemon.jsManager.WriteStoppedInstance(instanceID, instance)
require.NoError(t, err)
t.Cleanup(func() { _ = daemon.jsManager.DeleteStoppedInstance(instanceID) })
// Value holds decoded bytes (the gateway query parser decodes base64 from the CLI,
// then json.Marshal/Unmarshal round-trips []byte through base64 transparently)
newContent := "#!/bin/bash"
input := &ec2.ModifyInstanceAttributeInput{
InstanceId: aws.String(instanceID),
UserData: &ec2.BlobAttributeValue{Value: []byte(newContent)},
}
reqData, _ := json.Marshal(input)
reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
require.NoError(t, err)
assert.Equal(t, `{}`, string(reply.Data))
updated, err := daemon.jsManager.LoadStoppedInstance(instanceID)
require.NoError(t, err)
require.NotNil(t, updated)
assert.Equal(t, newContent, updated.UserData)
assert.Equal(t, "IyEvYmluL2Jhc2g=", *updated.RunInstancesInput.UserData)
}
func TestHandleEC2ModifyInstanceAttribute_SourceDestCheck(t *testing.T) {
natsURL := sharedJSNATSURL
daemon := createFullTestDaemonWithJetStream(t, natsURL)
sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", daemon.handleEC2ModifyInstanceAttribute)
require.NoError(t, err)
defer sub.Unsubscribe()
// SourceDestCheck is a no-op that succeeds without requiring a stopped instance
// in KV — Terraform sends this on running instances right after creation.
input := &ec2.ModifyInstanceAttributeInput{
InstanceId: aws.String("i-modify-sdc-001"),
SourceDestCheck: &ec2.AttributeBooleanValue{Value: aws.Bool(false)},
}
reqData, _ := json.Marshal(input)
reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
require.NoError(t, err)
assert.Equal(t, `{}`, string(reply.Data))
}
func TestHandleEC2ModifyInstanceAttribute_InstanceNotFound(t *testing.T) {
natsURL := sharedJSNATSURL
daemon := createFullTestDaemonWithJetStream(t, natsURL)
sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", daemon.handleEC2ModifyInstanceAttribute)
require.NoError(t, err)
defer sub.Unsubscribe()
input := &ec2.ModifyInstanceAttributeInput{
InstanceId: aws.String("i-nonexistent"),
InstanceType: &ec2.AttributeValue{Value: aws.String("t3.medium")},
}
reqData, _ := json.Marshal(input)
reply, err := daemon.natsConn.Request("ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
require.NoError(t, err)
var errResp map[string]any
err = json.Unmarshal(reply.Data, &errResp)
require.NoError(t, err)
assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, errResp["Code"])
}
func TestHandleEC2ModifyInstanceAttribute_NotStopped(t *testing.T) {
natsURL := sharedJSNATSURL
daemon := createFullTestDaemonWithJetStream(t, natsURL)
sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", daemon.handleEC2ModifyInstanceAttribute)
require.NoError(t, err)
defer sub.Unsubscribe()
instanceID := "i-modify-running-001"
instance := &vm.VM{
ID: instanceID,
Status: vm.StateRunning,
InstanceType: "t3.micro",
AccountID: testAccountID,
}
err = daemon.jsManager.WriteStoppedInstance(instanceID, instance)
require.NoError(t, err)
t.Cleanup(func() { _ = daemon.jsManager.DeleteStoppedInstance(instanceID) })
input := &ec2.ModifyInstanceAttributeInput{
InstanceId: aws.String(instanceID),
InstanceType: &ec2.AttributeValue{Value: aws.String("t3.medium")},
}
reqData, _ := json.Marshal(input)
reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
require.NoError(t, err)
var errResp map[string]any
err = json.Unmarshal(reply.Data, &errResp)
require.NoError(t, err)
assert.Equal(t, awserrors.ErrorIncorrectInstanceState, errResp["Code"])
}
func TestHandleEC2ModifyInstanceAttribute_ClearsStateReason(t *testing.T) {
natsURL := sharedJSNATSURL
daemon := createFullTestDaemonWithJetStream(t, natsURL)
sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", daemon.handleEC2ModifyInstanceAttribute)
require.NoError(t, err)
defer sub.Unsubscribe()
instanceID := "i-modify-recovery-001"
instance := &vm.VM{
ID: instanceID,
Status: vm.StateStopped,
InstanceType: "m7i.small",
AccountID: testAccountID,
Config: vm.Config{InstanceType: "m7i.small"},
Instance: &ec2.Instance{
InstanceId: aws.String(instanceID),
InstanceType: aws.String("m7i.small"),
StateReason: &ec2.StateReason{
Code: aws.String("Server.InsufficientInstanceCapacity"),
Message: aws.String("Instance type not available on any node"),
},
},
}
err = daemon.jsManager.WriteStoppedInstance(instanceID, instance)
require.NoError(t, err)
t.Cleanup(func() { _ = daemon.jsManager.DeleteStoppedInstance(instanceID) })
input := &ec2.ModifyInstanceAttributeInput{
InstanceId: aws.String(instanceID),
InstanceType: &ec2.AttributeValue{Value: aws.String("t3.micro")},
}
reqData, _ := json.Marshal(input)
reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
require.NoError(t, err)
assert.Equal(t, `{}`, string(reply.Data))
updated, err := daemon.jsManager.LoadStoppedInstance(instanceID)
require.NoError(t, err)
require.NotNil(t, updated)
assert.Equal(t, "t3.micro", updated.InstanceType)
assert.Equal(t, "t3.micro", updated.Config.InstanceType)
assert.Equal(t, "t3.micro", *updated.Instance.InstanceType)
assert.Nil(t, updated.Instance.StateReason)
}
func TestHandleEC2ModifyInstanceAttribute_InvalidTypeAccepted(t *testing.T) {
natsURL := sharedJSNATSURL
daemon := createFullTestDaemonWithJetStream(t, natsURL)
sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", daemon.handleEC2ModifyInstanceAttribute)
require.NoError(t, err)
defer sub.Unsubscribe()
instanceID := "i-modify-nonsense-001"
instance := &vm.VM{
ID: instanceID,
Status: vm.StateStopped,
InstanceType: "t3.micro",
AccountID: testAccountID,
Config: vm.Config{InstanceType: "t3.micro"},
Instance: &ec2.Instance{
InstanceId: aws.String(instanceID),
InstanceType: aws.String("t3.micro"),
},
}
err = daemon.jsManager.WriteStoppedInstance(instanceID, instance)
require.NoError(t, err)
t.Cleanup(func() { _ = daemon.jsManager.DeleteStoppedInstance(instanceID) })
// z99.mega is nonsense — modify does not pre-validate, matching AWS behavior
input := &ec2.ModifyInstanceAttributeInput{
InstanceId: aws.String(instanceID),
InstanceType: &ec2.AttributeValue{Value: aws.String("z99.mega")},
}
reqData, _ := json.Marshal(input)
reply, err := natsRequest(daemon.natsConn, "ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
require.NoError(t, err)
assert.Equal(t, `{}`, string(reply.Data))
updated, err := daemon.jsManager.LoadStoppedInstance(instanceID)
require.NoError(t, err)
require.NotNil(t, updated)
assert.Equal(t, "z99.mega", updated.InstanceType)
assert.Equal(t, "z99.mega", updated.Config.InstanceType)
assert.Equal(t, "z99.mega", *updated.Instance.InstanceType)
}
func TestHandleEC2ModifyInstanceAttribute_MissingInstanceID(t *testing.T) {
natsURL := sharedJSNATSURL
daemon := createFullTestDaemonWithJetStream(t, natsURL)
sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", daemon.handleEC2ModifyInstanceAttribute)
require.NoError(t, err)
defer sub.Unsubscribe()
input := &ec2.ModifyInstanceAttributeInput{}
reqData, _ := json.Marshal(input)
reply, err := daemon.natsConn.Request("ec2.ModifyInstanceAttribute", reqData, 5*time.Second)
require.NoError(t, err)
var errResp map[string]any
err = json.Unmarshal(reply.Data, &errResp)
require.NoError(t, err)
assert.Equal(t, awserrors.ErrorMissingParameter, errResp["Code"])
}
func TestHandleEC2ModifyInstanceAttribute_InvalidJSON(t *testing.T) {
natsURL := sharedJSNATSURL
daemon := createFullTestDaemonWithJetStream(t, natsURL)
sub, err := daemon.natsConn.QueueSubscribe("ec2.ModifyInstanceAttribute", "spinifex-workers", daemon.handleEC2ModifyInstanceAttribute)
require.NoError(t, err)
defer sub.Unsubscribe()
reply, err := daemon.natsConn.Request("ec2.ModifyInstanceAttribute", []byte(`{invalid`), 5*time.Second)
require.NoError(t, err)
var errResp map[string]any
err = json.Unmarshal(reply.Data, &errResp)
require.NoError(t, err)
assert.Equal(t, awserrors.ErrorServerInternal, errResp["Code"])
}
// --- DescribeInstanceAttribute daemon tests ---
@@ -3082,63 +2731,6 @@ func TestHandleEC2ModifyVolume_Success(t *testing.T) {
assert.Equal(t, "completed", *output.VolumeModification.ModificationState)
}
// --- handleEC2TerminateStoppedInstance with volumes ---
func TestHandleEC2TerminateStoppedInstance_WithVolumes(t *testing.T) {
daemon := createFullTestDaemonWithJetStream(t, sharedJSNATSURL)
// Subscribe a dummy ebs.delete handler
ebsDeleteSub, err := daemon.natsConn.Subscribe("ebs.delete", func(msg *nats.Msg) {
_ = msg.Respond([]byte(`{"status":"deleted"}`))
})
require.NoError(t, err)
defer ebsDeleteSub.Unsubscribe()
stoppedVM := &vm.VM{
ID: "i-term-vol-001",
Status: vm.StateStopped,
AccountID: testAccountID,
InstanceType: getTestInstanceType(t),
LastNode: "node-1",
Reservation: &ec2.Reservation{
ReservationId: aws.String("r-term-vol-001"),
OwnerId: aws.String("123456789012"),
},
Instance: &ec2.Instance{
InstanceId: aws.String("i-term-vol-001"),
InstanceType: aws.String(getTestInstanceType(t)),
},
}
// Add EFI, CloudInit, and a user volume with DeleteOnTermination
stoppedVM.EBSRequests.Requests = []types.EBSRequest{
{Name: "vol-efi-001", EFI: true},
{Name: "vol-ci-001", CloudInit: true},
{Name: "vol-user-001", DeleteOnTermination: true},
{Name: "vol-keep-001", DeleteOnTermination: false},
}
err = daemon.jsManager.WriteStoppedInstance(stoppedVM.ID, stoppedVM)
require.NoError(t, err)
sub, err := daemon.natsConn.QueueSubscribe("ec2.terminate", "spinifex-workers", daemon.handleEC2TerminateStoppedInstance)
require.NoError(t, err)
defer sub.Unsubscribe()
reqData, _ := json.Marshal(map[string]string{"instance_id": "i-term-vol-001"})
reply, err := natsRequest(daemon.natsConn, "ec2.terminate", reqData, 10*time.Second)
require.NoError(t, err)
var resp map[string]string
err = json.Unmarshal(reply.Data, &resp)
require.NoError(t, err)
assert.Equal(t, "terminated", resp["status"])
assert.Equal(t, "i-term-vol-001", resp["instanceId"])
loaded, err := daemon.jsManager.LoadStoppedInstance("i-term-vol-001")
require.NoError(t, err)
assert.Nil(t, loaded)
}
// --- handleEC2DescribeInstanceTypes with capacity filter ---
func TestHandleEC2DescribeInstanceTypes_CapacityFilter(t *testing.T) {
@@ -3887,7 +3479,7 @@ func TestHandleEC2TerminateStoppedInstance_WritesToTerminatedKV(t *testing.T) {
require.NoError(t, err)
defer sub.Unsubscribe()
reqData, _ := json.Marshal(terminateStoppedInstanceRequest{InstanceID: stoppedVM.ID})
reqData, _ := json.Marshal(handlers_ec2_instance.TerminateStoppedInstanceInput{InstanceID: stoppedVM.ID})
reply, err := natsRequest(daemon.natsConn, "ec2.terminate", reqData, 30*time.Second)
require.NoError(t, err)
assert.Contains(t, string(reply.Data), "terminated")
spinifex/daemon/daemon_system_instance.go
+1 -1
View File
@@ -225,7 +225,7 @@ func (d *Daemon) LaunchSystemInstance(input *handlers_elbv2.SystemInstanceInput)
vpcID = *result.NetworkInterfaces[0].VpcId
}
portName := "port-" + instance.ENIId
d.publishNATEvent("vpc.add-nat", vpcID, publicIP, privateIP, portName, instance.ENIMac)
utils.PublishNATEvent(d.natsConn, "vpc.add-nat", vpcID, publicIP, privateIP, portName, instance.ENIMac)
instance.PublicIP = publicIP
instance.PublicIPPool = poolName
slog.Info("LaunchSystemInstance: allocated public IP via direct IPAM",
spinifex/daemon/gpu_adapter.go
+26
View File
@@ -0,0 +1,26 @@
package daemon
import (
handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance"
)
// daemonGPUClaimer adapts the daemon's gpu.Manager + GPUModelOverrides config to
// the transport-neutral handlers_ec2_instance.GPUClaimer surface. Keeps the gpu
// package out of the instance service.
type daemonGPUClaimer struct {
d *Daemon
}
var _ handlers_ec2_instance.GPUClaimer = (*daemonGPUClaimer)(nil)
func (g *daemonGPUClaimer) Claim(instanceID string) (string, bool, error) {
dev, err := g.d.gpuManager.Claim(instanceID)
if err != nil {
return "", false, err
}
return dev.PCIAddress, gpuXVGAEnabled(dev, g.d.config.Daemon.GPUModelOverrides), nil
}
func (g *daemonGPUClaimer) Release(instanceID string) error {
return g.d.gpuManager.Release(instanceID)
}
spinifex/daemon/vm_adapters.go
+2 -1
View File
@@ -12,6 +12,7 @@ import (
"github.com/mulgadc/spinifex/spinifex/awserrors"
handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/vm"
"github.com/nats-io/nats.go"
)
@@ -570,7 +571,7 @@ func (a *instanceCleanerAdapter) ReleasePublicIP(instance *vm.VM) {
logicalIP = *instance.Instance.PrivateIpAddress
}
}
a.d.publishNATEvent("vpc.delete-nat", vpcId, instance.PublicIP, logicalIP, portName, "")
utils.PublishNATEvent(a.d.natsConn, "vpc.delete-nat", vpcId, instance.PublicIP, logicalIP, portName, "")
if err := a.d.externalIPAM.ReleaseIP(instance.PublicIPPool, instance.PublicIP); err != nil {
slog.Warn("Failed to release public IP on termination",
spinifex/daemon/vpc_adapter.go
+52
View File
@@ -0,0 +1,52 @@
package daemon
import (
"github.com/aws/aws-sdk-go/service/ec2"
handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance"
)
// daemonENICreator adapts the daemon's VPCServiceImpl to the
// handlers_ec2_instance.ENICreator interface so RunInstances can stay free of
// the wider VPC service surface. Returned subnet records are translated to
// handlers_ec2_instance.SubnetInfo to avoid a cyclic instance↔vpc import.
type daemonENICreator struct {
d *Daemon
}
var _ handlers_ec2_instance.ENICreator = (*daemonENICreator)(nil)
func (a *daemonENICreator) GetDefaultSubnet(accountID string) (*handlers_ec2_instance.SubnetInfo, error) {
rec, err := a.d.vpcService.GetDefaultSubnet(accountID)
if err != nil {
return nil, err
}
return &handlers_ec2_instance.SubnetInfo{
SubnetID: rec.SubnetId,
VpcID: rec.VpcId,
MapPublicIpOnLaunch: rec.MapPublicIpOnLaunch,
}, nil
}
func (a *daemonENICreator) GetSubnet(accountID, subnetID string) (*handlers_ec2_instance.SubnetInfo, error) {
rec, err := a.d.vpcService.GetSubnet(accountID, subnetID)
if err != nil {
return nil, err
}
return &handlers_ec2_instance.SubnetInfo{
SubnetID: rec.SubnetId,
VpcID: rec.VpcId,
MapPublicIpOnLaunch: rec.MapPublicIpOnLaunch,
}, nil
}
func (a *daemonENICreator) CreateNetworkInterface(input *ec2.CreateNetworkInterfaceInput, accountID string) (*ec2.CreateNetworkInterfaceOutput, error) {
return a.d.vpcService.CreateNetworkInterface(input, accountID)
}
func (a *daemonENICreator) AttachENI(accountID, eniID, instanceID string, deviceIndex int64) (string, error) {
return a.d.vpcService.AttachENI(accountID, eniID, instanceID, deviceIndex)
}
func (a *daemonENICreator) UpdateENIPublicIP(accountID, eniID, publicIP, poolName string) error {
return a.d.vpcService.UpdateENIPublicIP(accountID, eniID, publicIP, poolName)
}
spinifex/handlers/ec2/instance/service.go
+105 -2
View File
@@ -3,6 +3,7 @@ package handlers_ec2_instance
import (
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/vm"
"github.com/mulgadc/viperblock/viperblock"
)
// InstanceService defines the interface for EC2 instance operations business logic
@@ -13,6 +14,31 @@ type InstanceService interface {
DescribeInstanceAttribute(input *ec2.DescribeInstanceAttributeInput, accountID string) (*ec2.DescribeInstanceAttributeOutput, error)
DescribeStoppedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error)
DescribeTerminatedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error)
ModifyInstanceAttribute(input *ec2.ModifyInstanceAttributeInput, accountID string) (*ec2.ModifyInstanceAttributeOutput, error)
StartStoppedInstance(input *StartStoppedInstanceInput, accountID string) (*StartStoppedInstanceOutput, error)
TerminateStoppedInstance(input *TerminateStoppedInstanceInput, accountID string) (*TerminateStoppedInstanceOutput, error)
}
// StartStoppedInstanceInput is the payload for ec2.StartStoppedInstance.
type StartStoppedInstanceInput struct {
InstanceID string `json:"instance_id"`
}
// StartStoppedInstanceOutput is the response payload.
type StartStoppedInstanceOutput struct {
Status string `json:"status"`
InstanceID string `json:"instanceId"`
}
// TerminateStoppedInstanceInput is the payload for ec2.TerminateStoppedInstance.
type TerminateStoppedInstanceInput struct {
InstanceID string `json:"instance_id"`
}
// TerminateStoppedInstanceOutput is the response payload.
type TerminateStoppedInstanceOutput struct {
Status string `json:"status"`
InstanceID string `json:"instanceId"`
}
// ResourceCapacityProvider exposes per-node instance-type availability used by
@@ -21,10 +47,87 @@ type ResourceCapacityProvider interface {
GetAvailableInstanceTypeInfos(showCapacity bool) []*ec2.InstanceTypeInfo
}
// StoppedInstanceStore provides KV-backed read access to stopped/terminated
// instances. Implemented by daemon.JetStreamManager.
// InstanceTypeAllocator extends ResourceCapacityProvider with the mutating
// resource-reservation methods used by StartStoppedInstance. Implemented by
// daemon.ResourceManager.
type InstanceTypeAllocator interface {
ResourceCapacityProvider
Allocate(instanceType *ec2.InstanceTypeInfo) error
Deallocate(instanceType *ec2.InstanceTypeInfo)
CanAllocate(instanceType *ec2.InstanceTypeInfo, count int) int
InstanceTypes() map[string]*ec2.InstanceTypeInfo
}
// GPUClaimer binds a GPU (and its IOMMU group) to vfio-pci for a starting
// instance. Returns the PCI address and whether QEMU should set x-vga=on.
// nil claimer means the daemon has no GPU passthrough configured.
type GPUClaimer interface {
Claim(instanceID string) (pciAddress string, xvgaEnabled bool, err error)
Release(instanceID string) error
}
// StoppedInstanceStore covers KV-backed read/write access for stopped instances
// and read+write access for terminated instances. Implemented by
// daemon.JetStreamManager.
type StoppedInstanceStore interface {
LoadStoppedInstance(instanceID string) (*vm.VM, error)
ListStoppedInstances() ([]*vm.VM, error)
ListTerminatedInstances() ([]*vm.VM, error)
WriteStoppedInstance(instanceID string, instance *vm.VM) error
DeleteStoppedInstance(instanceID string) error
WriteTerminatedInstance(instanceID string, instance *vm.VM) error
}
// VolumeDeleter deletes EBS volumes. Implemented by handlers/ec2/volume's
// VolumeServiceImpl (used by the daemon).
type VolumeDeleter interface {
DeleteVolume(input *ec2.DeleteVolumeInput, accountID string) (*ec2.DeleteVolumeOutput, error)
}
// ENIDeleter deletes ENIs. Implemented by handlers/ec2/vpc's VPCServiceImpl.
type ENIDeleter interface {
DeleteNetworkInterface(input *ec2.DeleteNetworkInterfaceInput, accountID string) (*ec2.DeleteNetworkInterfaceOutput, error)
}
// PublicIPReleaser releases a previously allocated public IP back to a pool.
// Implemented by handlers/ec2/vpc.ExternalIPAM.
type PublicIPReleaser interface {
ReleaseIP(pool, ip string) error
}
// AMIMetaLoader resolves an AMI ID to its metadata for ownership/validation
// during RunInstances. Implemented by handlers/ec2/image.ImageServiceImpl.
type AMIMetaLoader interface {
GetAMIConfig(imageID string) (viperblock.AMIMetadata, error)
}
// KeyPairValidator checks that a named key pair exists for an account during
// RunInstances. Implemented by handlers/ec2/key.KeyServiceImpl.
type KeyPairValidator interface {
ValidateKeyPairExists(accountID, keyName string) error
}
// SubnetInfo carries the subset of subnet metadata RunInstances needs to
// resolve default subnets and decide whether to auto-assign a public IP.
type SubnetInfo struct {
SubnetID string
VpcID string
MapPublicIpOnLaunch bool
}
// ENICreator covers the VPC/ENI operations RunInstances performs while
// auto-attaching a primary interface. Implemented via an adapter over
// handlers/ec2/vpc.VPCServiceImpl on the daemon.
type ENICreator interface {
GetDefaultSubnet(accountID string) (*SubnetInfo, error)
GetSubnet(accountID, subnetID string) (*SubnetInfo, error)
CreateNetworkInterface(input *ec2.CreateNetworkInterfaceInput, accountID string) (*ec2.CreateNetworkInterfaceOutput, error)
AttachENI(accountID, eniID, instanceID string, deviceIndex int64) (string, error)
UpdateENIPublicIP(accountID, eniID, publicIP, poolName string) error
}
// PublicIPAllocator allocates a public IP to an instance/ENI from a pool.
// Implemented by handlers/ec2/vpc.ExternalIPAM.
type PublicIPAllocator interface {
AllocateIP(region, az, allocType, allocID, eniID, instanceID string) (publicIP, poolName string, err error)
}
spinifex/handlers/ec2/instance/service_impl.go
+791 -2
View File
@@ -3,6 +3,7 @@ package handlers_ec2_instance
import (
"bytes"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"io"
@@ -20,6 +21,7 @@ import (
"github.com/mulgadc/spinifex/spinifex/awserrors"
"github.com/mulgadc/spinifex/spinifex/config"
"github.com/mulgadc/spinifex/spinifex/filterutil"
"github.com/mulgadc/spinifex/spinifex/instancetypes"
"github.com/mulgadc/spinifex/spinifex/objectstore"
spxtypes "github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
@@ -234,8 +236,16 @@ type InstanceServiceImpl struct {
natsConn *nats.Conn
objectStore objectstore.ObjectStore
vmMgr *vm.Manager
resourceMgr ResourceCapacityProvider
resourceMgr InstanceTypeAllocator
stoppedStore StoppedInstanceStore
volumeDeleter VolumeDeleter
eniDeleter ENIDeleter
ipReleaser PublicIPReleaser
gpuClaimer GPUClaimer
amiLoader AMIMetaLoader
keyValidator KeyPairValidator
eniCreator ENICreator
ipAllocator PublicIPAllocator
}
// NewInstanceServiceImpl creates a new instance service implementation for daemon use
@@ -245,7 +255,7 @@ func NewInstanceServiceImpl(
nc *nats.Conn,
store objectstore.ObjectStore,
vmMgr *vm.Manager,
resourceMgr ResourceCapacityProvider,
resourceMgr InstanceTypeAllocator,
stoppedStore StoppedInstanceStore,
) *InstanceServiceImpl {
return &InstanceServiceImpl{
@@ -259,6 +269,31 @@ func NewInstanceServiceImpl(
}
}
// SetTerminationDeps wires the dependencies required by TerminateStoppedInstance.
// Kept separate from the main constructor so handlers needing only read or
// modify paths can construct a service without dragging the VPC/volume stack in.
func (s *InstanceServiceImpl) SetTerminationDeps(vd VolumeDeleter, ed ENIDeleter, pr PublicIPReleaser) {
s.volumeDeleter = vd
s.eniDeleter = ed
s.ipReleaser = pr
}
// SetGPUClaimer wires the GPU passthrough claim/release dependency used by
// StartStoppedInstance. nil disables GPU passthrough for the service.
func (s *InstanceServiceImpl) SetGPUClaimer(g GPUClaimer) {
s.gpuClaimer = g
}
// SetRunInstancesDeps wires the AMI/key/VPC/IPAM dependencies required by
// RunInstances. Kept separate from the constructor so handlers needing only
// read paths can build the service without dragging the full stack in.
func (s *InstanceServiceImpl) SetRunInstancesDeps(ami AMIMetaLoader, key KeyPairValidator, eni ENICreator, ipam PublicIPAllocator) {
s.amiLoader = ami
s.keyValidator = key
s.eniCreator = eni
s.ipAllocator = ipam
}
// RunInstance creates a single EC2 instance (called per-instance by daemon)
// Returns the VM struct and EC2 instance metadata
func (s *InstanceServiceImpl) RunInstance(input *ec2.RunInstancesInput) (*vm.VM, *ec2.Instance, error) {
@@ -300,6 +335,441 @@ func (s *InstanceServiceImpl) RunInstance(input *ec2.RunInstancesInput) (*vm.VM,
return instance, ec2Instance, nil
}
// PrepareRunInstances validates RunInstances input, allocates per-node capacity,
// creates per-instance VM + ec2.Instance metadata, auto-resolves the subnet,
// creates the primary ENI, and auto-assigns a public IP when the subnet has
// MapPublicIpOnLaunch. It does NOT touch vmMgr, WriteState, or NATS
// subscriptions — those are daemon concerns so the daemon can respond to AWS
// between Prepare and Launch (preserving the original respond-then-launch
// timing). Returns the reservation, the prepared VMs, and the instance-type
// info so the caller can deallocate on failure.
func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, []*vm.VM, *ec2.InstanceTypeInfo, error) {
if accountID == "" {
return nil, nil, nil, errors.New(awserrors.ErrorServerInternal)
}
if input == nil || input.InstanceType == nil {
return nil, nil, nil, errors.New(awserrors.ErrorMissingParameter)
}
instanceType, exists := s.instanceTypes[*input.InstanceType]
if !exists {
slog.Error("PrepareRunInstances: invalid instance type", "InstanceType", *input.InstanceType)
return nil, nil, nil, errors.New(awserrors.ErrorInvalidInstanceType)
}
if input.ImageId == nil || *input.ImageId == "" {
slog.Error("PrepareRunInstances: missing ImageId")
return nil, nil, nil, errors.New(awserrors.ErrorMissingParameter)
}
if s.amiLoader == nil {
slog.Error("PrepareRunInstances: AMI loader not initialized")
return nil, nil, nil, errors.New(awserrors.ErrorServerInternal)
}
amiMeta, err := s.amiLoader.GetAMIConfig(*input.ImageId)
if err != nil {
slog.Error("PrepareRunInstances: AMI not found", "imageId", *input.ImageId, "err", err)
return nil, nil, nil, errors.New(awserrors.ErrorInvalidAMIIDNotFound)
}
// Caller must own the AMI or it must carry a non-account-ID owner alias
// (e.g. "self", "spinifex", "") indicating a system/legacy AMI.
amiOwner := amiMeta.ImageOwnerAlias
if amiOwner != "" && amiOwner != accountID && utils.IsAccountID(amiOwner) {
slog.Warn("PrepareRunInstances: AMI not owned by caller", "imageId", *input.ImageId, "amiOwner", amiOwner, "accountID", accountID)
return nil, nil, nil, errors.New(awserrors.ErrorInvalidAMIIDNotFound)
}
if input.KeyName != nil && *input.KeyName != "" {
if s.keyValidator == nil {
slog.Error("PrepareRunInstances: key validator not initialized")
return nil, nil, nil, errors.New(awserrors.ErrorServerInternal)
}
if err := s.keyValidator.ValidateKeyPairExists(accountID, *input.KeyName); err != nil {
slog.Error("PrepareRunInstances: key pair not found", "keyName", *input.KeyName, "err", err)
return nil, nil, nil, errors.New(awserrors.ErrorInvalidKeyPairNotFound)
}
}
minCount := int(*input.MinCount)
maxCount := int(*input.MaxCount)
allocatableCount := s.resourceMgr.CanAllocate(instanceType, maxCount)
if allocatableCount < minCount {
slog.Error("PrepareRunInstances: insufficient capacity", "requested", minCount, "available", allocatableCount, "InstanceType", *input.InstanceType)
return nil, nil, nil, errors.New(awserrors.ErrorInsufficientInstanceCapacity)
}
launchCount := allocatableCount
slog.Info("PrepareRunInstances: count determined", "min", minCount, "max", maxCount, "launching", launchCount)
var allocatedCount int
for i := 0; i < launchCount; i++ {
if err := s.resourceMgr.Allocate(instanceType); err != nil {
slog.Error("PrepareRunInstances: allocate failed mid-allocation", "allocated", allocatedCount, "err", err)
break
}
allocatedCount++
}
if allocatedCount < minCount {
for i := 0; i < allocatedCount; i++ {
s.resourceMgr.Deallocate(instanceType)
}
slog.Error("PrepareRunInstances: insufficient capacity after allocation", "allocated", allocatedCount, "minCount", minCount)
return nil, nil, nil, errors.New(awserrors.ErrorInsufficientInstanceCapacity)
}
launchCount = allocatedCount
var instances []*vm.VM
var allEC2Instances []*ec2.Instance
var lastRunErr error
for i := 0; i < launchCount; i++ {
instance, ec2Instance, err := s.RunInstance(input)
if err != nil {
slog.Error("PrepareRunInstances: RunInstance failed", "index", i, "err", err)
lastRunErr = err
s.resourceMgr.Deallocate(instanceType)
continue
}
// Terraform with associate_public_ip_address sends subnet/SG via
// NetworkInterfaces[0]; lift to top-level so the rest works uniformly.
if (input.SubnetId == nil || *input.SubnetId == "") &&
len(input.NetworkInterfaces) > 0 && input.NetworkInterfaces[0] != nil {
nic := input.NetworkInterfaces[0]
if nic.SubnetId != nil && *nic.SubnetId != "" {
input.SubnetId = nic.SubnetId
}
if len(input.SecurityGroupIds) == 0 && len(nic.Groups) > 0 {
input.SecurityGroupIds = nic.Groups
}
}
if (input.SubnetId == nil || *input.SubnetId == "") && s.eniCreator != nil {
defaultSubnet, dsErr := s.eniCreator.GetDefaultSubnet(accountID)
if dsErr == nil && defaultSubnet != nil {
input.SubnetId = aws.String(defaultSubnet.SubnetID)
slog.Info("PrepareRunInstances: resolved default subnet", "instanceId", instance.ID, "subnetId", defaultSubnet.SubnetID)
}
}
if input.SubnetId != nil && *input.SubnetId != "" && s.eniCreator != nil {
eniOut, eniErr := s.eniCreator.CreateNetworkInterface(&ec2.CreateNetworkInterfaceInput{
SubnetId: input.SubnetId,
Description: aws.String("Primary network interface for " + instance.ID),
}, accountID)
if eniErr != nil {
slog.Error("PrepareRunInstances: auto-create ENI failed", "instanceId", instance.ID, "subnetId", *input.SubnetId, "err", eniErr)
lastRunErr = eniErr
s.resourceMgr.Deallocate(instanceType)
continue
}
eni := eniOut.NetworkInterface
instance.ENIId = *eni.NetworkInterfaceId
instance.ENIMac = *eni.MacAddress
if _, attachErr := s.eniCreator.AttachENI(accountID, instance.ENIId, instance.ID, 0); attachErr != nil {
slog.Error("PrepareRunInstances: failed to attach ENI to instance record — ELBv2 target IP resolution will fail", "eniId", instance.ENIId, "instanceId", instance.ID, "err", attachErr)
}
ec2Instance.SetPrivateIpAddress(*eni.PrivateIpAddress)
ec2Instance.SetSubnetId(*input.SubnetId)
ec2Instance.SetVpcId(*eni.VpcId)
ec2Instance.NetworkInterfaces = []*ec2.InstanceNetworkInterface{
{
NetworkInterfaceId: eni.NetworkInterfaceId,
PrivateIpAddress: eni.PrivateIpAddress,
MacAddress: eni.MacAddress,
SubnetId: input.SubnetId,
VpcId: eni.VpcId,
Status: aws.String("in-use"),
Attachment: &ec2.InstanceNetworkInterfaceAttachment{
DeviceIndex: aws.Int64(0),
Status: aws.String("attached"),
},
},
}
slog.Info("PrepareRunInstances: auto-created ENI for VPC instance",
"instanceId", instance.ID,
"eniId", instance.ENIId,
"privateIp", *eni.PrivateIpAddress,
"mac", instance.ENIMac,
)
if s.ipAllocator != nil {
subnet, subErr := s.eniCreator.GetSubnet(accountID, *input.SubnetId)
if subErr == nil && subnet != nil && subnet.MapPublicIpOnLaunch {
region := s.config.Region
az := s.config.AZ
publicIP, poolName, allocErr := s.ipAllocator.AllocateIP(region, az, "auto_assign", "", *eni.NetworkInterfaceId, instance.ID)
if allocErr != nil {
slog.Warn("PrepareRunInstances: failed to allocate public IP", "instanceId", instance.ID, "err", allocErr)
} else {
if updateErr := s.eniCreator.UpdateENIPublicIP(accountID, *eni.NetworkInterfaceId, publicIP, poolName); updateErr != nil {
slog.Warn("PrepareRunInstances: failed to update ENI with public IP", "eniId", *eni.NetworkInterfaceId, "err", updateErr)
}
portName := "port-" + *eni.NetworkInterfaceId
utils.PublishNATEvent(s.natsConn, "vpc.add-nat", *eni.VpcId, publicIP, *eni.PrivateIpAddress, portName, *eni.MacAddress)
ec2Instance.PublicIpAddress = aws.String(publicIP)
instance.PublicIP = publicIP
instance.PublicIPPool = poolName
slog.Info("PrepareRunInstances: auto-assigned public IP",
"instanceId", instance.ID,
"publicIp", publicIP,
"privateIp", *eni.PrivateIpAddress,
"pool", poolName,
)
}
}
}
}
instances = append(instances, instance)
allEC2Instances = append(allEC2Instances, ec2Instance)
}
if len(instances) < minCount {
for range instances {
s.resourceMgr.Deallocate(instanceType)
}
errCode := awserrors.ErrorServerInternal
if lastRunErr != nil {
if _, ok := awserrors.ErrorLookup[lastRunErr.Error()]; ok {
errCode = lastRunErr.Error()
}
}
slog.Error("PrepareRunInstances: failed to create minimum instances", "created", len(instances), "minCount", minCount, "err", errCode)
return nil, nil, nil, errors.New(errCode)
}
reservation := &ec2.Reservation{}
reservation.SetReservationId(utils.GenerateResourceID("r"))
reservation.SetOwnerId(accountID)
reservation.Instances = allEC2Instances
for _, instance := range instances {
instance.Reservation = reservation
instance.AccountID = accountID
if input.Placement != nil && input.Placement.GroupName != nil && *input.Placement.GroupName != "" {
instance.PlacementGroupName = *input.Placement.GroupName
}
}
return reservation, instances, instanceType, nil
}
// LaunchRunInstances takes the VMs created by PrepareRunInstances (already
// inserted into vmMgr by the caller) and runs the heavyweight launch loop:
// volume preparation, GPU claim, vmMgr.Run. Failures mark the VM as failed
// and deallocate resources but do not abort the loop — partial success
// matches AWS behaviour where some instances in a reservation may fail.
func (s *InstanceServiceImpl) LaunchRunInstances(instances []*vm.VM, input *ec2.RunInstancesInput, instanceType *ec2.InstanceTypeInfo) {
var successCount int
for _, instance := range instances {
// Skip if a concurrent request terminated the instance during prepare.
status := s.vmMgr.Status(instance)
if status != vm.StatePending && status != vm.StateProvisioning {
slog.Info("LaunchRunInstances: instance state changed during provisioning, skipping launch",
"instanceId", instance.ID, "status", string(status))
continue
}
// Pre-compute dev MAC so cloud-init can generate per-interface netplan
// that suppresses the default route on the dev/hostfwd NIC.
if s.config.Daemon.DevNetworking && instance.ENIId != "" {
instance.DevMAC = vm.GenerateDevMAC(instance.ID)
}
volumeInfos, err := s.GenerateVolumes(input, instance)
if err != nil {
slog.Error("LaunchRunInstances: GenerateVolumes failed", "instanceId", instance.ID, "err", err)
s.vmMgr.MarkFailed(instance, "volume_preparation_failed")
continue
}
instance.Instance.BlockDeviceMappings = make([]*ec2.InstanceBlockDeviceMapping, 0, len(volumeInfos))
for _, vi := range volumeInfos {
mapping := &ec2.InstanceBlockDeviceMapping{}
mapping.SetDeviceName(vi.DeviceName)
mapping.Ebs = &ec2.EbsInstanceBlockDevice{}
mapping.Ebs.SetVolumeId(vi.VolumeId)
mapping.Ebs.SetAttachTime(vi.AttachTime)
mapping.Ebs.SetDeleteOnTermination(vi.DeleteOnTermination)
mapping.Ebs.SetStatus("attached")
instance.Instance.BlockDeviceMappings = append(instance.Instance.BlockDeviceMappings, mapping)
}
if s.gpuClaimer != nil && instancetypes.IsGPUType(instanceType) {
pciAddr, xvga, gpuErr := s.gpuClaimer.Claim(instance.ID)
if gpuErr != nil {
slog.Error("LaunchRunInstances: GPU claim failed", "instanceId", instance.ID, "err", gpuErr)
s.vmMgr.MarkFailed(instance, "gpu_claim_failed")
continue
}
instance.GPUPCIAddress = pciAddr
instance.GPUXVGAEnabled = xvga
slog.Info("LaunchRunInstances: GPU claimed for instance", "instanceId", instance.ID, "gpu", pciAddr, "xvga", xvga)
}
if err := s.vmMgr.Run(instance); err != nil {
slog.Error("LaunchRunInstances: vmMgr.Run failed", "instanceId", instance.ID, "err", err)
if instance.GPUPCIAddress != "" && s.gpuClaimer != nil {
if releaseErr := s.gpuClaimer.Release(instance.ID); releaseErr != nil {
slog.Error("LaunchRunInstances: GPU release failed after launch failure",
"instanceId", instance.ID, "err", releaseErr)
}
}
s.vmMgr.MarkFailed(instance, "launch_failed")
continue
}
s.vmMgr.UpdateGuestDeviceNames(instance)
successCount++
slog.Info("LaunchRunInstances: launched instance", "instanceId", instance.ID)
}
slog.Info("LaunchRunInstances: completed", "requested", len(instances), "launched", successCount)
}
// RunInstances satisfies the InstanceService interface for non-daemon callers
// (tests, mocks). The daemon's NATS handler bypasses this method and calls
// PrepareRunInstances + LaunchRunInstances directly so it can respond to AWS
// between the two phases, preserving the original respond-then-launch timing.
func (s *InstanceServiceImpl) RunInstances(input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, error) {
reservation, instances, instanceType, err := s.PrepareRunInstances(input, accountID)
if err != nil {
return nil, err
}
for _, instance := range instances {
s.vmMgr.Insert(instance)
}
s.LaunchRunInstances(instances, input, instanceType)
return reservation, nil
}
// RebootInstance handles an ec2.cmd reboot for an instance already running on
// this node. Returns nil on success; the dispatching daemon handler maps the
// returned error to a NATS error response and responds with `{}` on nil.
func (s *InstanceServiceImpl) RebootInstance(instance *vm.VM, command spxtypes.EC2InstanceCommand) error {
slog.Info("RebootInstance: rebooting instance", "id", command.ID)
if err := s.vmMgr.Reboot(instance.ID); err != nil {
switch {
case errors.Is(err, vm.ErrInstanceNotFound):
return errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
case errors.Is(err, vm.ErrInvalidTransition):
slog.Error("RebootInstance: instance not in running state",
"instanceId", command.ID, "err", err)
return errors.New(awserrors.ErrorIncorrectInstanceState)
default:
slog.Error("RebootInstance: reboot failed", "instanceId", command.ID, "err", err)
return errors.New(awserrors.ErrorServerInternal)
}
}
slog.Info("RebootInstance: rebooted", "instanceId", command.ID)
return nil
}
// StartInstance handles an ec2.cmd start for an instance that is stopped on
// this node (local vmMgr state, not the cross-node shared-KV path). Returns
// nil on success; the dispatching daemon handler responds with the running
// status payload.
func (s *InstanceServiceImpl) StartInstance(instance *vm.VM, command spxtypes.EC2InstanceCommand) error {
slog.Info("StartInstance: starting instance", "id", command.ID)
status := s.vmMgr.Status(instance)
if status != vm.StateStopped {
slog.Error("StartInstance: instance not in stopped state", "instanceId", command.ID, "status", status)
return errors.New(awserrors.ErrorIncorrectInstanceState)
}
instanceType, ok := s.resourceMgr.InstanceTypes()[instance.InstanceType]
if ok {
if err := s.resourceMgr.Allocate(instanceType); err != nil {
slog.Error("StartInstance: failed to allocate resources", "id", command.ID, "err", err)
return errors.New(awserrors.ErrorInsufficientInstanceCapacity)
}
}
// Clear stop attribute before launch so WriteState inside the manager
// persists the correct attributes. Without this, a daemon restart after
// a stop→start cycle would see StopInstance=true and skip reconnecting QEMU.
s.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Attributes = command.Attributes })
if err := s.vmMgr.Start(instance.ID); err != nil {
slog.Error("StartInstance: vmMgr.Start failed", "err", err)
if ok {
s.resourceMgr.Deallocate(instanceType)
}
return errors.New(awserrors.ErrorServerInternal)
}
s.vmMgr.UpdateGuestDeviceNames(instance)
slog.Info("StartInstance: started", "instanceId", instance.ID)
return nil
}
// StopOrTerminateInstance handles an ec2.cmd stop or terminate for an
// instance running on this node. Validates the transition synchronously,
// stamps the command attributes onto the VM, then dispatches the actual
// Stop/Terminate in a background goroutine so the daemon can ack immediately.
func (s *InstanceServiceImpl) StopOrTerminateInstance(instance *vm.VM, command spxtypes.EC2InstanceCommand) error {
isTerminate := command.Attributes.TerminateInstance
action := "Stopping"
initialState := vm.StateStopping
if isTerminate {
action = "Terminating"
initialState = vm.StateShuttingDown
}
slog.Info("StopOrTerminateInstance: "+action, "id", command.ID)
currentState := s.vmMgr.Status(instance)
// Idempotent: a concurrent terminate goroutine is already cleaning up.
if isTerminate && currentState == vm.StateShuttingDown {
slog.Info("StopOrTerminateInstance: instance already shutting down, terminate is idempotent", "instanceId", instance.ID)
return nil
}
// Validate the transition synchronously before dispatching so the AWS
// SDK sees IncorrectInstanceState (400) instead of a stale 200. The
// async Stop/Terminate path re-validates and surfaces vm.ErrInvalidTransition
// on a racing transition; we map that into the same AWS error below.
if !vm.IsValidTransition(currentState, initialState) {
slog.Warn("StopOrTerminateInstance: instance in incorrect state for "+strings.ToLower(action),
"instanceId", instance.ID, "currentState", string(currentState))
return errors.New(awserrors.ErrorIncorrectInstanceState)
}
// Stamp command attributes onto the VM before dispatch so the persisted
// state reflects the user-stop / user-terminate intent (e.g. for the
// recovery path that distinguishes user-stopped from crash-stopped).
s.vmMgr.UpdateState(instance.ID, func(v *vm.VM) { v.Attributes = command.Attributes })
go func(id string) {
var err error
if isTerminate {
err = s.vmMgr.Terminate(id)
} else {
err = s.vmMgr.Stop(id)
}
if err != nil {
if errors.Is(err, vm.ErrInvalidTransition) {
slog.Warn("StopOrTerminateInstance: lifecycle transition raced; ack already sent",
"id", id, "action", strings.ToLower(action), "err", err)
return
}
slog.Error("StopOrTerminateInstance: failed to "+strings.ToLower(action), "err", err, "id", id)
}
}(instance.ID)
return nil
}
func (s *InstanceServiceImpl) GenerateVolumes(input *ec2.RunInstancesInput, instance *vm.VM) ([]VolumeInfo, error) {
p := parseVolumeParams(input)
@@ -1157,6 +1627,325 @@ func (s *InstanceServiceImpl) describeInstancesFromKV(input *ec2.DescribeInstanc
return &ec2.DescribeInstancesOutput{Reservations: reservations}, nil
}
// ModifyInstanceAttribute applies a single attribute change to a stopped instance.
// All supported attributes (InstanceType, UserData) require the instance to be
// stopped. SourceDestCheck is a networking concept that does not apply to
// bare-metal VMs; it is accepted as a no-op on any instance state so Terraform
// and the AWS CLI do not error out.
func (s *InstanceServiceImpl) ModifyInstanceAttribute(input *ec2.ModifyInstanceAttributeInput, accountID string) (*ec2.ModifyInstanceAttributeOutput, error) {
if input.InstanceId == nil || *input.InstanceId == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
instanceID := *input.InstanceId
if input.SourceDestCheck != nil {
slog.Info("ModifyInstanceAttribute: accepting SourceDestCheck (no-op on bare metal)", "instanceId", instanceID)
return &ec2.ModifyInstanceAttributeOutput{}, nil
}
if s.stoppedStore == nil {
slog.Error("ModifyInstanceAttribute: stopped store not available")
return nil, errors.New(awserrors.ErrorServerInternal)
}
instance, err := s.stoppedStore.LoadStoppedInstance(instanceID)
if err != nil {
slog.Error("ModifyInstanceAttribute: failed to load stopped instance", "instanceId", instanceID, "err", err)
return nil, errors.New(awserrors.ErrorServerInternal)
}
if instance == nil {
slog.Warn("ModifyInstanceAttribute: instance not found in shared KV", "instanceId", instanceID)
return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
}
if instance.Status != vm.StateStopped {
slog.Error("ModifyInstanceAttribute: instance not in stopped state", "instanceId", instanceID, "status", instance.Status)
return nil, errors.New(awserrors.ErrorIncorrectInstanceState)
}
if !IsInstanceVisible(accountID, instance.AccountID) {
slog.Warn("ModifyInstanceAttribute: instance not visible",
"instanceId", instanceID, "callerAccount", accountID, "ownerAccount", instance.AccountID)
return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
}
if input.InstanceType != nil && input.InstanceType.Value != nil {
newType := *input.InstanceType.Value
if newType == "" {
slog.Error("ModifyInstanceAttribute: empty instance type value", "instanceId", instanceID)
return nil, errors.New(awserrors.ErrorInvalidInstanceAttributeValue)
}
if instance.Instance == nil {
slog.Error("ModifyInstanceAttribute: instance.Instance is nil, data integrity issue", "instanceId", instanceID)
return nil, errors.New(awserrors.ErrorServerInternal)
}
slog.Info("ModifyInstanceAttribute: changing instance type",
"instanceId", instanceID, "oldType", instance.InstanceType, "newType", newType)
instance.InstanceType = newType
instance.Config.InstanceType = newType
instance.Instance.InstanceType = aws.String(newType)
// Clear StateReason — resolves capacity-unavailable state from instance-type-missing bug.
instance.Instance.StateReason = nil
}
if input.UserData != nil && input.UserData.Value != nil {
slog.Info("ModifyInstanceAttribute: changing user data", "instanceId", instanceID)
instance.UserData = string(input.UserData.Value)
if instance.RunInstancesInput != nil {
instance.RunInstancesInput.UserData = aws.String(base64.StdEncoding.EncodeToString(input.UserData.Value))
}
}
if err := s.stoppedStore.WriteStoppedInstance(instanceID, instance); err != nil {
slog.Error("ModifyInstanceAttribute: failed to write modified instance to KV",
"instanceId", instanceID, "err", err)
return nil, errors.New(awserrors.ErrorServerInternal)
}
slog.Info("ModifyInstanceAttribute: completed successfully", "instanceId", instanceID)
return &ec2.ModifyInstanceAttributeOutput{}, nil
}
// StartStoppedInstance picks up a stopped instance from shared KV, re-launches
// it on this daemon node, then removes it from shared KV.
func (s *InstanceServiceImpl) StartStoppedInstance(input *StartStoppedInstanceInput, accountID string) (*StartStoppedInstanceOutput, error) {
if input.InstanceID == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
if s.stoppedStore == nil {
slog.Error("StartStoppedInstance: stopped store not available")
return nil, errors.New(awserrors.ErrorServerInternal)
}
if s.resourceMgr == nil {
slog.Error("StartStoppedInstance: resource manager not available")
return nil, errors.New(awserrors.ErrorServerInternal)
}
if s.vmMgr == nil {
slog.Error("StartStoppedInstance: vm manager not available")
return nil, errors.New(awserrors.ErrorServerInternal)
}
instance, err := s.stoppedStore.LoadStoppedInstance(input.InstanceID)
if err != nil {
slog.Error("StartStoppedInstance: failed to load stopped instance", "instanceId", input.InstanceID, "err", err)
return nil, errors.New(awserrors.ErrorServerInternal)
}
if instance == nil {
slog.Warn("StartStoppedInstance: instance not found in shared KV", "instanceId", input.InstanceID)
return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
}
if instance.Status != vm.StateStopped {
slog.Error("StartStoppedInstance: instance not in stopped state", "instanceId", input.InstanceID, "status", instance.Status)
return nil, errors.New(awserrors.ErrorIncorrectInstanceState)
}
if !IsInstanceVisible(accountID, instance.AccountID) {
slog.Warn("StartStoppedInstance: instance not visible",
"instanceId", input.InstanceID, "callerAccount", accountID, "ownerAccount", instance.AccountID)
return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
}
// Reset node-local fields that are stale after cross-node migration.
instance.ResetNodeLocalState()
instanceType, ok := s.resourceMgr.InstanceTypes()[instance.InstanceType]
if !ok {
slog.Error("StartStoppedInstance: instance type not available on this node",
"instanceId", input.InstanceID, "instanceType", instance.InstanceType)
return nil, errors.New(awserrors.ErrorInsufficientInstanceCapacity)
}
if err := s.resourceMgr.Allocate(instanceType); err != nil {
slog.Error("StartStoppedInstance: failed to allocate resources", "instanceId", input.InstanceID, "err", err)
return nil, errors.New(awserrors.ErrorInsufficientInstanceCapacity)
}
// Add to local map + clear stop attribute before launch.
instance.Attributes = spxtypes.EC2CommandAttributes{StartInstance: true}
s.vmMgr.Insert(instance)
// Claim GPU for GPU instance types — binds the full IOMMU group to vfio-pci.
gpuClaimed := false
if s.gpuClaimer != nil && instancetypes.IsGPUType(instanceType) {
pciAddr, xvga, gpuErr := s.gpuClaimer.Claim(instance.ID)
if gpuErr != nil {
slog.Error("StartStoppedInstance: GPU claim failed", "instanceId", input.InstanceID, "err", gpuErr)
s.resourceMgr.Deallocate(instanceType)
s.vmMgr.Delete(instance.ID)
return nil, errors.New(awserrors.ErrorInsufficientInstanceCapacity)
}
instance.GPUPCIAddress = pciAddr
instance.GPUXVGAEnabled = xvga
gpuClaimed = true
slog.Info("GPU claimed for instance", "instanceId", input.InstanceID, "gpu", pciAddr, "xvga", xvga)
}
if err := s.vmMgr.Run(instance); err != nil {
slog.Error("StartStoppedInstance: vmMgr.Run failed", "instanceId", input.InstanceID, "err", err)
if gpuClaimed {
if relErr := s.gpuClaimer.Release(instance.ID); relErr != nil {
slog.Error("StartStoppedInstance: GPU release failed after launch failure",
"instanceId", input.InstanceID, "err", relErr)
}
}
s.resourceMgr.Deallocate(instanceType)
s.vmMgr.Delete(instance.ID)
return nil, errors.New(awserrors.ErrorServerInternal)
}
// Discover actual guest device names via QMP query-block.
s.vmMgr.UpdateGuestDeviceNames(instance)
// Remove from shared KV now that it's running locally. Retry once on failure —
// a stale KV entry risks duplicate starts.
if err := s.stoppedStore.DeleteStoppedInstance(input.InstanceID); err != nil {
slog.Warn("StartStoppedInstance: first KV delete failed, retrying",
"instanceId", input.InstanceID, "err", err)
if retryErr := s.stoppedStore.DeleteStoppedInstance(input.InstanceID); retryErr != nil {
slog.Error("StartStoppedInstance: KV delete failed after retry, instance is running locally but stale entry remains in shared KV",
"instanceId", input.InstanceID, "err", retryErr)
}
}
slog.Info("Started stopped instance from shared KV", "instanceId", instance.ID)
return &StartStoppedInstanceOutput{Status: "running", InstanceID: instance.ID}, nil
}
// TerminateStoppedInstance picks up a stopped instance from shared KV, deletes
// its volumes, releases its public IP and ENI, writes it to the terminated
// bucket, then removes it from the stopped bucket. No QEMU shutdown or unmount
// is needed — the instance was already drained during stop.
func (s *InstanceServiceImpl) TerminateStoppedInstance(input *TerminateStoppedInstanceInput, accountID string) (*TerminateStoppedInstanceOutput, error) {
if input.InstanceID == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
if s.stoppedStore == nil {
slog.Error("TerminateStoppedInstance: stopped store not available")
return nil, errors.New(awserrors.ErrorServerInternal)
}
instance, err := s.stoppedStore.LoadStoppedInstance(input.InstanceID)
if err != nil {
slog.Error("TerminateStoppedInstance: failed to load stopped instance", "instanceId", input.InstanceID, "err", err)
return nil, errors.New(awserrors.ErrorServerInternal)
}
if instance == nil {
slog.Warn("TerminateStoppedInstance: instance not found in shared KV", "instanceId", input.InstanceID)
return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
}
if instance.Status != vm.StateStopped {
slog.Error("TerminateStoppedInstance: instance not in stopped state", "instanceId", input.InstanceID, "status", instance.Status)
return nil, errors.New(awserrors.ErrorIncorrectInstanceState)
}
if !IsInstanceVisible(accountID, instance.AccountID) {
slog.Warn("TerminateStoppedInstance: instance not visible",
"instanceId", input.InstanceID, "callerAccount", accountID, "ownerAccount", instance.AccountID)
return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
}
s.deleteInstanceVolumes(instance, input.InstanceID)
s.releaseInstancePublicIP(instance, input.InstanceID)
s.deleteInstanceENI(instance, input.InstanceID)
// Write to terminated KV FIRST so the instance is visible in DescribeInstances.
// If this fails the instance stays in the stopped bucket — safe to retry.
instance.Status = vm.StateTerminated
if err := s.stoppedStore.WriteTerminatedInstance(input.InstanceID, instance); err != nil {
slog.Error("TerminateStoppedInstance: failed to write to terminated KV, aborting", "instanceId", input.InstanceID, "err", err)
return nil, errors.New(awserrors.ErrorServerInternal)
}
// Now safe to remove from stopped KV. Retry once on failure so the instance
// doesn't appear in both buckets.
if err := s.stoppedStore.DeleteStoppedInstance(input.InstanceID); err != nil {
slog.Warn("TerminateStoppedInstance: first stopped KV delete failed, retrying",
"instanceId", input.InstanceID, "err", err)
if retryErr := s.stoppedStore.DeleteStoppedInstance(input.InstanceID); retryErr != nil {
slog.Error("TerminateStoppedInstance: stopped KV delete failed after retry, instance may appear in both buckets",
"instanceId", input.InstanceID, "err", retryErr)
}
}
slog.Info("Terminated stopped instance from shared KV", "instanceId", input.InstanceID)
return &TerminateStoppedInstanceOutput{Status: "terminated", InstanceID: input.InstanceID}, nil
}
func (s *InstanceServiceImpl) deleteInstanceVolumes(instance *vm.VM, instanceID string) {
instance.EBSRequests.Mu.Lock()
defer instance.EBSRequests.Mu.Unlock()
for _, ebsRequest := range instance.EBSRequests.Requests {
// Internal volumes (EFI, cloud-init) are always cleaned up via ebs.delete.
if ebsRequest.EFI || ebsRequest.CloudInit {
ebsDeleteData, err := json.Marshal(spxtypes.EBSDeleteRequest{Volume: ebsRequest.Name})
if err != nil {
slog.Error("TerminateStoppedInstance: failed to marshal ebs.delete request", "name", ebsRequest.Name, "err", err)
continue
}
deleteMsg, err := s.natsConn.Request("ebs.delete", ebsDeleteData, 30*time.Second)
if err != nil {
slog.Warn("TerminateStoppedInstance: ebs.delete failed for internal volume", "name", ebsRequest.Name, "err", err)
} else {
slog.Info("TerminateStoppedInstance: ebs.delete sent for internal volume", "name", ebsRequest.Name, "data", string(deleteMsg.Data))
}
continue
}
// User-visible volumes: respect DeleteOnTermination.
if !ebsRequest.DeleteOnTermination {
slog.Info("TerminateStoppedInstance: volume has DeleteOnTermination=false, skipping", "name", ebsRequest.Name)
continue
}
slog.Info("TerminateStoppedInstance: deleting volume with DeleteOnTermination=true", "name", ebsRequest.Name)
if s.volumeDeleter == nil {
slog.Warn("TerminateStoppedInstance: volume deleter not configured, skipping", "name", ebsRequest.Name)
continue
}
if _, err := s.volumeDeleter.DeleteVolume(&ec2.DeleteVolumeInput{
VolumeId: &ebsRequest.Name,
}, instance.AccountID); err != nil {
slog.Error("TerminateStoppedInstance: failed to delete volume", "name", ebsRequest.Name, "err", err)
}
}
_ = instanceID
}
func (s *InstanceServiceImpl) releaseInstancePublicIP(instance *vm.VM, instanceID string) {
if instance.PublicIP == "" || instance.PublicIPPool == "" || s.ipReleaser == nil {
return
}
portName := "port-" + instance.ENIId
vpcID := ""
logicalIP := ""
if instance.Instance != nil {
if instance.Instance.VpcId != nil {
vpcID = *instance.Instance.VpcId
}
if instance.Instance.PrivateIpAddress != nil {
logicalIP = *instance.Instance.PrivateIpAddress
}
}
utils.PublishNATEvent(s.natsConn, "vpc.delete-nat", vpcID, instance.PublicIP, logicalIP, portName, "")
if err := s.ipReleaser.ReleaseIP(instance.PublicIPPool, instance.PublicIP); err != nil {
slog.Warn("TerminateStoppedInstance: failed to release public IP", "ip", instance.PublicIP, "pool", instance.PublicIPPool, "err", err)
} else {
slog.Info("TerminateStoppedInstance: released public IP", "ip", instance.PublicIP, "instanceId", instanceID)
}
}
func (s *InstanceServiceImpl) deleteInstanceENI(instance *vm.VM, instanceID string) {
if instance.ENIId == "" || s.eniDeleter == nil {
return
}
if _, err := s.eniDeleter.DeleteNetworkInterface(&ec2.DeleteNetworkInterfaceInput{
NetworkInterfaceId: &instance.ENIId,
}, instance.AccountID); err != nil {
slog.Error("TerminateStoppedInstance: failed to delete ENI", "eni", instance.ENIId, "err", err)
} else {
slog.Info("TerminateStoppedInstance: deleted ENI", "eni", instance.ENIId, "instanceId", instanceID)
}
}
// DescribeInstanceAttribute returns a single requested attribute for an instance.
// Checks running instances first (in-memory), then falls back to stopped instances in KV.
func (s *InstanceServiceImpl) DescribeInstanceAttribute(input *ec2.DescribeInstanceAttributeInput, accountID string) (*ec2.DescribeInstanceAttributeOutput, error) {
spinifex/handlers/ec2/instance/service_impl_test.go
+1264 -9
View File
File diff suppressed because it is too large Load Diff
spinifex/handlers/ec2/instance/service_mock.go
+12
View File
@@ -55,3 +55,15 @@ func (s *MockInstanceService) DescribeStoppedInstances(_ *ec2.DescribeInstancesI
func (s *MockInstanceService) DescribeTerminatedInstances(_ *ec2.DescribeInstancesInput, _ string) (*ec2.DescribeInstancesOutput, error) {
return &ec2.DescribeInstancesOutput{}, nil
}
func (s *MockInstanceService) ModifyInstanceAttribute(_ *ec2.ModifyInstanceAttributeInput, _ string) (*ec2.ModifyInstanceAttributeOutput, error) {
return &ec2.ModifyInstanceAttributeOutput{}, nil
}
func (s *MockInstanceService) StartStoppedInstance(input *StartStoppedInstanceInput, _ string) (*StartStoppedInstanceOutput, error) {
return &StartStoppedInstanceOutput{Status: "running", InstanceID: input.InstanceID}, nil
}
func (s *MockInstanceService) TerminateStoppedInstance(input *TerminateStoppedInstanceInput, _ string) (*TerminateStoppedInstanceOutput, error) {
return &TerminateStoppedInstanceOutput{Status: "terminated", InstanceID: input.InstanceID}, nil
}
spinifex/handlers/ec2/instance/service_nats.go
+12
View File
@@ -60,3 +60,15 @@ func (s *NATSInstanceService) DescribeStoppedInstances(input *ec2.DescribeInstan
func (s *NATSInstanceService) DescribeTerminatedInstances(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) {
return utils.NATSRequest[ec2.DescribeInstancesOutput](s.natsConn, "ec2.DescribeTerminatedInstances", input, 30*time.Second, accountID)
}
func (s *NATSInstanceService) ModifyInstanceAttribute(input *ec2.ModifyInstanceAttributeInput, accountID string) (*ec2.ModifyInstanceAttributeOutput, error) {
return utils.NATSRequest[ec2.ModifyInstanceAttributeOutput](s.natsConn, "ec2.ModifyInstanceAttribute", input, 30*time.Second, accountID)
}
func (s *NATSInstanceService) StartStoppedInstance(input *StartStoppedInstanceInput, accountID string) (*StartStoppedInstanceOutput, error) {
return utils.NATSRequest[StartStoppedInstanceOutput](s.natsConn, "ec2.StartStoppedInstance", input, 30*time.Second, accountID)
}
func (s *NATSInstanceService) TerminateStoppedInstance(input *TerminateStoppedInstanceInput, accountID string) (*TerminateStoppedInstanceOutput, error) {
return utils.NATSRequest[TerminateStoppedInstanceOutput](s.natsConn, "ec2.TerminateStoppedInstance", input, 30*time.Second, accountID)
}
spinifex/utils/nats.go
+22
View File
@@ -382,6 +382,28 @@ func PublishEvent(nc *nats.Conn, topic string, event any) {
}
}
// PublishNATEvent sends a NAT lifecycle event (vpc.add-nat or vpc.delete-nat)
// to NATS. vpc.add-nat uses request-reply to commit the OVN NAT rule before
// returning (prevents ARP propagation races); vpc.delete-nat is fire-and-forget.
func PublishNATEvent(nc *nats.Conn, topic, vpcID, externalIP, logicalIP, portName, mac string) {
evt := struct {
VpcId string `json:"vpc_id"`
ExternalIP string `json:"external_ip"`
LogicalIP string `json:"logical_ip"`
PortName string `json:"port_name"`
MAC string `json:"mac"`
}{VpcId: vpcID, ExternalIP: externalIP, LogicalIP: logicalIP, PortName: portName, MAC: mac}
if topic == "vpc.add-nat" {
if err := RequestEvent(nc, topic, evt, 10*time.Second); err != nil {
slog.Warn("PublishNATEvent: failed to add NAT rule — OVN dnat_and_snat rule not created; restart vpcd or re-associate EIP to recover",
"topic", topic, "externalIP", externalIP, "logicalIP", logicalIP, "err", err)
}
return
}
PublishEvent(nc, topic, evt)
}
// RequestEvent marshals event as JSON and sends a NATS request, waiting for a
// response. This ensures the subscriber has processed the event before the
// caller continues. Returns an error if the request times out or the responder