diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index e5ff82ad0..1fbdd9a36 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -1096,6 +1096,10 @@ 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) + d.accountService, err = initServiceWithRetry("account settings service", func() (*handlers_ec2_account.AccountSettingsServiceImpl, error) { return handlers_ec2_account.NewAccountSettingsServiceImplWithNATS(d.config, d.natsConn) }) diff --git a/spinifex/daemon/daemon_handlers_instance.go b/spinifex/daemon/daemon_handlers_instance.go index 6983b9b39..37b6e9d88 100644 --- a/spinifex/daemon/daemon_handlers_instance.go +++ b/spinifex/daemon/daemon_handlers_instance.go @@ -6,7 +6,6 @@ import ( "fmt" "log/slog" "strings" - "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" @@ -248,7 +247,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { } // 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) + utils.PublishNATEvent(d.natsConn, "vpc.add-nat", *eni.VpcId, publicIP, *eni.PrivateIpAddress, portName, *eni.MacAddress) // Set on ec2Instance response ec2Instance.PublicIpAddress = aws.String(publicIP) instance.PublicIP = publicIP @@ -904,151 +903,8 @@ func (d *Daemon) handleEC2StartStoppedInstance(msg *nats.Msg) { } } -// 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. @@ -1277,26 +1133,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) -} diff --git a/spinifex/daemon/daemon_handlers_instance_test.go b/spinifex/daemon/daemon_handlers_instance_test.go index b6e238875..1fe6d96f1 100644 --- a/spinifex/daemon/daemon_handlers_instance_test.go +++ b/spinifex/daemon/daemon_handlers_instance_test.go @@ -257,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"]) } @@ -265,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"]) } @@ -279,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"]) @@ -302,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 @@ -329,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 @@ -347,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"]) diff --git a/spinifex/daemon/daemon_handlers_test.go b/spinifex/daemon/daemon_handlers_test.go index 173b9d91d..651b1e441 100644 --- a/spinifex/daemon/daemon_handlers_test.go +++ b/spinifex/daemon/daemon_handlers_test.go @@ -1363,121 +1363,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 @@ -2833,63 +2725,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) { @@ -3638,7 +3473,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") diff --git a/spinifex/daemon/daemon_system_instance.go b/spinifex/daemon/daemon_system_instance.go index 5926bd687..68b041949 100644 --- a/spinifex/daemon/daemon_system_instance.go +++ b/spinifex/daemon/daemon_system_instance.go @@ -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", diff --git a/spinifex/daemon/vm_adapters.go b/spinifex/daemon/vm_adapters.go index bb2d9b42b..970a4152e 100644 --- a/spinifex/daemon/vm_adapters.go +++ b/spinifex/daemon/vm_adapters.go @@ -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", diff --git a/spinifex/handlers/ec2/instance/service.go b/spinifex/handlers/ec2/instance/service.go index 5f4baef06..52b824d6b 100644 --- a/spinifex/handlers/ec2/instance/service.go +++ b/spinifex/handlers/ec2/instance/service.go @@ -14,6 +14,18 @@ type InstanceService interface { 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) + TerminateStoppedInstance(input *TerminateStoppedInstanceInput, accountID string) (*TerminateStoppedInstanceOutput, error) +} + +// 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 @@ -22,11 +34,31 @@ type ResourceCapacityProvider interface { GetAvailableInstanceTypeInfos(showCapacity bool) []*ec2.InstanceTypeInfo } -// StoppedInstanceStore provides KV-backed read/write access to stopped and -// read access to terminated instances. Implemented by daemon.JetStreamManager. +// 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 } diff --git a/spinifex/handlers/ec2/instance/service_impl.go b/spinifex/handlers/ec2/instance/service_impl.go index 7754ba445..e74ce2539 100644 --- a/spinifex/handlers/ec2/instance/service_impl.go +++ b/spinifex/handlers/ec2/instance/service_impl.go @@ -3,6 +3,7 @@ package handlers_ec2_instance import ( "bytes" "encoding/base64" + "encoding/json" "errors" "fmt" "io" @@ -236,6 +237,9 @@ type InstanceServiceImpl struct { vmMgr *vm.Manager resourceMgr ResourceCapacityProvider stoppedStore StoppedInstanceStore + volumeDeleter VolumeDeleter + eniDeleter ENIDeleter + ipReleaser PublicIPReleaser } // NewInstanceServiceImpl creates a new instance service implementation for daemon use @@ -259,6 +263,15 @@ 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 +} + // 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) { @@ -1237,6 +1250,142 @@ func (s *InstanceServiceImpl) ModifyInstanceAttribute(input *ec2.ModifyInstanceA return &ec2.ModifyInstanceAttributeOutput{}, 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) { diff --git a/spinifex/handlers/ec2/instance/service_impl_test.go b/spinifex/handlers/ec2/instance/service_impl_test.go index af93a14ee..cf3bad0b3 100644 --- a/spinifex/handlers/ec2/instance/service_impl_test.go +++ b/spinifex/handlers/ec2/instance/service_impl_test.go @@ -2,18 +2,24 @@ package handlers_ec2_instance import ( "bytes" + "encoding/json" + "errors" "fmt" "strings" "testing" "text/template" + "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/config" "github.com/mulgadc/spinifex/spinifex/objectstore" + spxtypes "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/mulgadc/spinifex/spinifex/vm" + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -653,14 +659,20 @@ func (f *fakeResourceCapacityProvider) GetAvailableInstanceTypeInfos(showCapacit } type fakeStoppedStore struct { - stopped []*vm.VM - terminated []*vm.VM - loadByID map[string]*vm.VM - wroteStopped map[string]*vm.VM - listErr error - listTermErr error - loadErr error - writeErr error + stopped []*vm.VM + terminated []*vm.VM + loadByID map[string]*vm.VM + wroteStopped map[string]*vm.VM + wroteTerminated map[string]*vm.VM + deletedStopped []string + listErr error + listTermErr error + loadErr error + writeErr error + writeTermErr error + deleteErr error + deleteFailFirst bool + deleteAttempts int } func (f *fakeStoppedStore) LoadStoppedInstance(id string) (*vm.VM, error) { @@ -694,6 +706,27 @@ func (f *fakeStoppedStore) WriteStoppedInstance(id string, instance *vm.VM) erro f.wroteStopped[id] = instance return nil } +func (f *fakeStoppedStore) WriteTerminatedInstance(id string, instance *vm.VM) error { + if f.writeTermErr != nil { + return f.writeTermErr + } + if f.wroteTerminated == nil { + f.wroteTerminated = make(map[string]*vm.VM) + } + f.wroteTerminated[id] = instance + return nil +} +func (f *fakeStoppedStore) DeleteStoppedInstance(id string) error { + f.deleteAttempts++ + if f.deleteFailFirst && f.deleteAttempts == 1 { + return errors.New("transient delete failure") + } + if f.deleteErr != nil { + return f.deleteErr + } + f.deletedStopped = append(f.deletedStopped, id) + return nil +} func TestDescribeInstanceTypes_NilResourceMgr(t *testing.T) { svc := &InstanceServiceImpl{} @@ -1267,3 +1300,266 @@ func TestModifyInstanceAttribute_WriteError(t *testing.T) { require.Error(t, err) assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) } + +// --- TerminateStoppedInstance tests --- + +type fakeVolumeDeleter struct { + calls []string + deleted []string + err error +} + +func (f *fakeVolumeDeleter) DeleteVolume(input *ec2.DeleteVolumeInput, _ string) (*ec2.DeleteVolumeOutput, error) { + id := aws.StringValue(input.VolumeId) + f.calls = append(f.calls, id) + if f.err != nil { + return nil, f.err + } + f.deleted = append(f.deleted, id) + return &ec2.DeleteVolumeOutput{}, nil +} + +type fakeENIDeleter struct { + calls []string + err error +} + +func (f *fakeENIDeleter) DeleteNetworkInterface(input *ec2.DeleteNetworkInterfaceInput, _ string) (*ec2.DeleteNetworkInterfaceOutput, error) { + f.calls = append(f.calls, aws.StringValue(input.NetworkInterfaceId)) + if f.err != nil { + return nil, f.err + } + return &ec2.DeleteNetworkInterfaceOutput{}, nil +} + +type fakePublicIPReleaser struct { + pool string + ip string + err error +} + +func (f *fakePublicIPReleaser) ReleaseIP(pool, ip string) error { + f.pool = pool + f.ip = ip + return f.err +} + +// embeddedNATS spins up an in-process NATS server scoped to the test and +// returns a connected client. Used for service tests that exercise the +// ebs.delete path inside TerminateStoppedInstance. +func embeddedNATS(t *testing.T) *nats.Conn { + t.Helper() + opts := &server.Options{ + Host: "127.0.0.1", + Port: -1, + NoLog: true, + NoSigs: true, + } + ns, err := server.NewServer(opts) + require.NoError(t, err) + go ns.Start() + require.True(t, ns.ReadyForConnections(5*time.Second)) + t.Cleanup(func() { ns.Shutdown() }) + + nc, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + t.Cleanup(func() { nc.Close() }) + return nc +} + +func TestTerminateStoppedInstance_MissingInstanceID(t *testing.T) { + svc := &InstanceServiceImpl{} + _, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{}, "acc") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorMissingParameter, err.Error()) +} + +func TestTerminateStoppedInstance_NilStore(t *testing.T) { + svc := &InstanceServiceImpl{} + _, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: "i-1"}, "acc") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) +} + +func TestTerminateStoppedInstance_LoadError(t *testing.T) { + store := &fakeStoppedStore{loadErr: errors.New("kv down")} + svc := &InstanceServiceImpl{stoppedStore: store} + + _, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: "i-1"}, "acc") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) +} + +func TestTerminateStoppedInstance_NotFound(t *testing.T) { + store := &fakeStoppedStore{loadByID: map[string]*vm.VM{}} + svc := &InstanceServiceImpl{stoppedStore: store} + + _, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: "i-missing"}, "acc") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error()) +} + +func TestTerminateStoppedInstance_NotStopped(t *testing.T) { + id := "i-running" + store := &fakeStoppedStore{loadByID: map[string]*vm.VM{ + id: {ID: id, Status: vm.StateRunning, AccountID: "acc"}, + }} + svc := &InstanceServiceImpl{stoppedStore: store} + + _, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorIncorrectInstanceState, err.Error()) +} + +func TestTerminateStoppedInstance_NotVisible(t *testing.T) { + id := "i-stopped" + store := &fakeStoppedStore{loadByID: map[string]*vm.VM{ + id: {ID: id, Status: vm.StateStopped, AccountID: "owner-acc"}, + }} + svc := &InstanceServiceImpl{stoppedStore: store} + + _, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "other-acc") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error()) +} + +func TestTerminateStoppedInstance_HappyPath(t *testing.T) { + id := "i-term-001" + store := &fakeStoppedStore{loadByID: map[string]*vm.VM{ + id: {ID: id, Status: vm.StateStopped, AccountID: "acc"}, + }} + svc := &InstanceServiceImpl{stoppedStore: store} + + out, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc") + require.NoError(t, err) + require.NotNil(t, out) + assert.Equal(t, "terminated", out.Status) + assert.Equal(t, id, out.InstanceID) + + require.NotNil(t, store.wroteTerminated[id]) + assert.Equal(t, vm.StateTerminated, store.wroteTerminated[id].Status) + assert.Contains(t, store.deletedStopped, id) +} + +func TestTerminateStoppedInstance_WriteTerminatedError_Aborts(t *testing.T) { + id := "i-werr" + store := &fakeStoppedStore{ + loadByID: map[string]*vm.VM{id: {ID: id, Status: vm.StateStopped, AccountID: "acc"}}, + writeTermErr: errors.New("kv write boom"), + } + svc := &InstanceServiceImpl{stoppedStore: store} + + _, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) + assert.Empty(t, store.deletedStopped, "stopped delete must not run when terminated write fails") +} + +func TestTerminateStoppedInstance_RetriesStoppedDelete(t *testing.T) { + id := "i-retry" + store := &fakeStoppedStore{ + loadByID: map[string]*vm.VM{id: {ID: id, Status: vm.StateStopped, AccountID: "acc"}}, + deleteFailFirst: true, + } + svc := &InstanceServiceImpl{stoppedStore: store} + + _, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc") + require.NoError(t, err) + assert.Equal(t, 2, store.deleteAttempts, "first delete fails, retry succeeds") + assert.Contains(t, store.deletedStopped, id) +} + +func TestTerminateStoppedInstance_UserVolumeDeleted(t *testing.T) { + id := "i-vol" + v := &vm.VM{ID: id, Status: vm.StateStopped, AccountID: "acc"} + v.EBSRequests.Requests = []spxtypes.EBSRequest{ + {Name: "vol-user-001", DeleteOnTermination: true}, + {Name: "vol-keep-001", DeleteOnTermination: false}, + } + store := &fakeStoppedStore{loadByID: map[string]*vm.VM{id: v}} + vd := &fakeVolumeDeleter{} + svc := &InstanceServiceImpl{stoppedStore: store, volumeDeleter: vd} + + _, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc") + require.NoError(t, err) + assert.Equal(t, []string{"vol-user-001"}, vd.calls, "only DeleteOnTermination=true volumes deleted") + assert.Equal(t, []string{"vol-user-001"}, vd.deleted) +} + +func TestTerminateStoppedInstance_NoVolumeDeleterSkipsGracefully(t *testing.T) { + id := "i-no-vd" + v := &vm.VM{ID: id, Status: vm.StateStopped, AccountID: "acc"} + v.EBSRequests.Requests = []spxtypes.EBSRequest{ + {Name: "vol-user-001", DeleteOnTermination: true}, + } + store := &fakeStoppedStore{loadByID: map[string]*vm.VM{id: v}} + svc := &InstanceServiceImpl{stoppedStore: store} + + _, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc") + require.NoError(t, err, "missing VolumeDeleter must not abort termination") + require.NotNil(t, store.wroteTerminated[id]) +} + +func TestTerminateStoppedInstance_InternalVolumesViaNATS(t *testing.T) { + id := "i-int-vol" + v := &vm.VM{ID: id, Status: vm.StateStopped, AccountID: "acc"} + v.EBSRequests.Requests = []spxtypes.EBSRequest{ + {Name: "vol-efi-001", EFI: true}, + {Name: "vol-ci-001", CloudInit: true}, + } + store := &fakeStoppedStore{loadByID: map[string]*vm.VM{id: v}} + + nc := embeddedNATS(t) + var ebsDeleted []string + sub, err := nc.Subscribe("ebs.delete", func(msg *nats.Msg) { + var req spxtypes.EBSDeleteRequest + _ = json.Unmarshal(msg.Data, &req) + ebsDeleted = append(ebsDeleted, req.Volume) + _ = msg.Respond([]byte(`{"Success":true}`)) + }) + require.NoError(t, err) + defer sub.Unsubscribe() + + svc := &InstanceServiceImpl{stoppedStore: store, natsConn: nc} + + _, err = svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc") + require.NoError(t, err) + assert.ElementsMatch(t, []string{"vol-efi-001", "vol-ci-001"}, ebsDeleted) +} + +func TestTerminateStoppedInstance_PublicIPReleased(t *testing.T) { + id := "i-pubip" + v := &vm.VM{ + ID: id, + Status: vm.StateStopped, + AccountID: "acc", + PublicIP: "203.0.113.5", + PublicIPPool: "pool-a", + ENIId: "eni-xyz", + Instance: &ec2.Instance{ + InstanceId: aws.String(id), + VpcId: aws.String("vpc-1"), + PrivateIpAddress: aws.String("10.0.0.5"), + }, + } + store := &fakeStoppedStore{loadByID: map[string]*vm.VM{id: v}} + pr := &fakePublicIPReleaser{} + svc := &InstanceServiceImpl{stoppedStore: store, ipReleaser: pr} + + _, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc") + require.NoError(t, err) + assert.Equal(t, "pool-a", pr.pool) + assert.Equal(t, "203.0.113.5", pr.ip) +} + +func TestTerminateStoppedInstance_ENIDeleted(t *testing.T) { + id := "i-eni" + v := &vm.VM{ID: id, Status: vm.StateStopped, AccountID: "acc", ENIId: "eni-1234"} + store := &fakeStoppedStore{loadByID: map[string]*vm.VM{id: v}} + ed := &fakeENIDeleter{} + svc := &InstanceServiceImpl{stoppedStore: store, eniDeleter: ed} + + _, err := svc.TerminateStoppedInstance(&TerminateStoppedInstanceInput{InstanceID: id}, "acc") + require.NoError(t, err) + assert.Equal(t, []string{"eni-1234"}, ed.calls) +} diff --git a/spinifex/handlers/ec2/instance/service_mock.go b/spinifex/handlers/ec2/instance/service_mock.go index 203688552..dfae7fd7b 100644 --- a/spinifex/handlers/ec2/instance/service_mock.go +++ b/spinifex/handlers/ec2/instance/service_mock.go @@ -59,3 +59,7 @@ func (s *MockInstanceService) DescribeTerminatedInstances(_ *ec2.DescribeInstanc func (s *MockInstanceService) ModifyInstanceAttribute(_ *ec2.ModifyInstanceAttributeInput, _ string) (*ec2.ModifyInstanceAttributeOutput, error) { return &ec2.ModifyInstanceAttributeOutput{}, nil } + +func (s *MockInstanceService) TerminateStoppedInstance(input *TerminateStoppedInstanceInput, _ string) (*TerminateStoppedInstanceOutput, error) { + return &TerminateStoppedInstanceOutput{Status: "terminated", InstanceID: input.InstanceID}, nil +} diff --git a/spinifex/handlers/ec2/instance/service_nats.go b/spinifex/handlers/ec2/instance/service_nats.go index a09d6ac45..4540f7fdf 100644 --- a/spinifex/handlers/ec2/instance/service_nats.go +++ b/spinifex/handlers/ec2/instance/service_nats.go @@ -64,3 +64,7 @@ func (s *NATSInstanceService) DescribeTerminatedInstances(input *ec2.DescribeIns 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) TerminateStoppedInstance(input *TerminateStoppedInstanceInput, accountID string) (*TerminateStoppedInstanceOutput, error) { + return utils.NATSRequest[TerminateStoppedInstanceOutput](s.natsConn, "ec2.TerminateStoppedInstance", input, 30*time.Second, accountID) +} diff --git a/spinifex/utils/nats.go b/spinifex/utils/nats.go index c5df9d79f..7cedebd4a 100644 --- a/spinifex/utils/nats.go +++ b/spinifex/utils/nats.go @@ -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