feat: disable api termination (#224)

This commit is contained in:
Julian Sommer
2026-05-14 14:35:00 +10:00
committed by GitHub
parent 8841241541
commit f7efc69344
11 changed files with 477 additions and 135 deletions
+3 -3
View File
@@ -142,12 +142,12 @@ file, deferred to a later phase.
| `describe-instances` | `--instance-ids`, `--filters` (instance-state-name, instance-id, instance-type, vpc-id, subnet-id, tag:\*, tag-key, tag-value) | `--max-results`, `--next-token`, `--dry-run` | None | Gateway fans out NATS `ec2.DescribeInstances` to all nodes (no queue group) → each daemon returns local instances → gateway aggregates and returns combined list. Filters applied per-node before aggregation (reduces payload). Also applies to stopped/terminated instances via `describeInstancesFromKV()`. | 1. Describe all instances (no filter)<br>2. Describe by instance ID<br>3. Describe with filters (e.g. instance-state-name)<br>4. Instance not found returns empty set<br>5. Multi-node aggregation returns instances from all nodes<br>6. Filter by tag<br>7. Unknown filter returns InvalidParameterValue | **DONE** |
| `start-instances` | `--instance-ids` | `--dry-run`, `--force` | `run-instances` (instance must exist in stopped state) | Gateway sends NATS `ec2.cmd.{instance-id}` → daemon restarts stopped QEMU process with same config → state transitions stopped→pending→running | 1. Start a stopped instance<br>2. Start already-running instance (error: IncorrectInstanceState)<br>3. Start with invalid instance ID<br>4. Verify volumes re-mount on start | **DONE** |
| `stop-instances` | `--instance-ids` | `--force`, `--hibernate`, `--dry-run` | `run-instances` (instance must be running) | Gateway sends NATS to target node → daemon issues QMP `system_powerdown` for graceful shutdown → monitors heartbeat until QEMU exits → state transitions running→stopping→stopped | 1. Graceful stop of running instance<br>2. Force stop (kills QEMU process)<br>3. Stop already-stopped instance (error)<br>4. Verify ~30s heartbeat detection | **DONE** |
| `terminate-instances` | `--instance-ids`, `DeleteOnTermination` (per-volume flag, default true) | `--dry-run` | `run-instances` (instance must exist) | Gateway sends NATS to target node → daemon kills QEMU process → cleans up NBD mounts → deletes volumes with `DeleteOnTermination=true` via `volumeService.DeleteVolume()` (S3 cleanup of vol/, vol-efi/, vol-cloudinit/) → internal volumes (EFI, cloud-init) always cleaned up via `ebs.delete` NATS → volumes with `DeleteOnTermination=false` left in available state → state→terminated | 1. Terminate running instance<br>2. Terminate stopped instance<br>3. Terminate with DeleteOnTermination=true deletes volumes<br>4. Terminate with DeleteOnTermination=false preserves volumes<br>5. Terminate already-terminated (idempotent)<br>6. Internal volumes (EFI, cloud-init) always cleaned up<br>7. Invalid instance ID | **DONE** |
| `terminate-instances` | `--instance-ids`, `DeleteOnTermination` (per-volume flag, default true) | `--dry-run` | `run-instances` (instance must exist) | Gateway sends NATS to target node → daemon checks `DisableApiTermination` first and rejects with `OperationNotPermitted` if set (running or stopped, no side effects) → kills QEMU process → cleans up NBD mounts → deletes volumes with `DeleteOnTermination=true` via `volumeService.DeleteVolume()` (S3 cleanup of vol/, vol-efi/, vol-cloudinit/) → internal volumes (EFI, cloud-init) always cleaned up via `ebs.delete` NATS → volumes with `DeleteOnTermination=false` left in available state → state→terminated | 1. Terminate running instance<br>2. Terminate stopped instance<br>3. Terminate with DeleteOnTermination=true deletes volumes<br>4. Terminate with DeleteOnTermination=false preserves volumes<br>5. Terminate already-terminated (idempotent)<br>6. Internal volumes (EFI, cloud-init) always cleaned up<br>7. Invalid instance ID<br>8. Running protected instance rejected (OperationNotPermitted, state stays running)<br>9. Stopped protected instance rejected (resources intact) | **DONE** |
| `reboot-instances` | `--instance-ids` | `--dry-run` | `run-instances` (instance must be running) | Gateway validates instance IDs → sends EC2InstanceCommand with `RebootInstance=true` via NATS `ec2.cmd.{instanceId}` → daemon validates instance is in StateRunning (returns IncorrectInstanceState if stopped) → issues QMP `system_reset` → instance reboots without stopping, stays in running state | 1. Reboot running instance<br>2. Reboot multiple instances<br>3. Reboot stopped instance (error: IncorrectInstanceState)<br>4. Instance not found (error: InvalidInstanceID.NotFound)<br>5. Verify instance stays in running state after reboot | **DONE** |
| `describe-instance-types` | `--filters` (capacity filter only) | `--instance-types`, `--max-results`, `--next-token`, `--dry-run`, all other filters | None | Gateway fans out NATS `ec2.DescribeInstanceTypes` to all nodes → each daemon reports supported types (t3.micro/small/medium/large) with vCPU/memory specs → gateway deduplicates and returns | 1. List all instance types<br>2. Filter by specific type<br>3. Filter with `capacity=true` shows available slots<br>4. Verify vCPU/memory specs match hardware | **DONE** |
| `modify-instance-attribute` | `--instance-id`, `--instance-type`, `--user-data` | `--disable-api-termination`, `--ebs-optimized`, `--source-dest-check`, `--instance-initiated-shutdown-behavior`, `--block-device-mappings`, `--groups`, `--ena-support`, `--sriov-net-support` | Instance must be stopped (in NATS KV) | Gateway validates input (exactly one attribute per call, instance ID format) → NATS `ec2.ModifyInstanceAttribute` with `spinifex-workers` queue group → daemon loads stopped instance from JetStream KV → applies attribute change → writes back to KV → returns `{}` on success. **InstanceType**: updates vm.InstanceType, Config, and Instance fields; clears StateReason (enables recovery from instance-type-missing bug). **UserData**: stores decoded content in vm.UserData and re-encodes to base64 for RunInstancesInput (cloud-init on next start). No instance type pre-validation (matches AWS — invalid types accepted, fail at StartInstances time). | 1. Change instance type while stopped<br>2. Change user data while stopped<br>3. Modify running instance (error: NotFound — running instances not in KV)<br>4. Instance not found (error: InvalidInstanceID.NotFound)<br>5. Instance not stopped (error: IncorrectInstanceState)<br>6. Invalid instance type accepted (fails on start with InsufficientInstanceCapacity)<br>7. StateReason cleared on type change (recovery from capacity-unavailable)<br>8. Missing/malformed instance ID (error: InvalidInstanceID.Malformed)<br>9. No attribute set (error: InvalidParameterValue)<br>10. Multiple attributes in one call (error: InvalidParameterValue) | **DONE** |
| `modify-instance-attribute` | `--instance-id`, `--instance-type`, `--user-data`, `--disable-api-termination` | `--ebs-optimized`, `--source-dest-check`, `--instance-initiated-shutdown-behavior`, `--block-device-mappings`, `--groups`, `--ena-support`, `--sriov-net-support` | Instance must exist; `instance-type` and `user-data` require stopped state, `disable-api-termination` works on running or stopped | Gateway validates input (exactly one attribute per call, instance ID format) → NATS `ec2.ModifyInstanceAttribute` with `spinifex-workers` queue group → daemon dispatches by state: stopped instances loaded from JetStream KV, running instances updated via `vm.Manager.UpdateAndPersist` (atomic flag check + persist under one lock to prevent Modify/Terminate race) → returns `{}` on success. **InstanceType**: updates vm.InstanceType, Config, and Instance fields; clears StateReason (enables recovery from instance-type-missing bug). **UserData**: stores decoded content in vm.UserData and re-encodes to base64 for RunInstancesInput (cloud-init on next start). **DisableApiTermination**: stamps `RunInstancesInput.DisableApiTermination`; flag survives stop/start and daemon restarts. No instance type pre-validation (matches AWS — invalid types accepted, fail at StartInstances time). | 1. Change instance type while stopped<br>2. Change user data while stopped<br>3. Toggle disable-api-termination on running instance<br>4. Toggle disable-api-termination on stopped instance<br>5. Modify instance-type on running instance (error: NotFound — running instances not in KV)<br>6. Instance not found (error: InvalidInstanceID.NotFound)<br>7. Instance not stopped for type/user-data change (error: IncorrectInstanceState)<br>8. Invalid instance type accepted (fails on start with InsufficientInstanceCapacity)<br>9. StateReason cleared on type change (recovery from capacity-unavailable)<br>10. Missing/malformed instance ID (error: InvalidInstanceID.Malformed)<br>11. No attribute set (error: InvalidParameterValue)<br>12. Multiple attributes in one call (error: InvalidParameterValue) | **DONE** |
| `get-console-output` | `--instance-id` | `--latest` (always returns latest), `--dry-run` | Instance must be running on a node | Gateway sends NATS `ec2.{instanceId}.GetConsoleOutput` (per-instance topic, routed to owning node) → daemon reads console log file from disk → returns last 64KB base64-encoded with timestamp. Always available regardless of serial console access setting (matches AWS behavior). | 1. Get output from running instance<br>2. Empty log file returns empty output<br>3. Instance not found (error: InvalidInstanceID.NotFound) | **DONE** |
| `describe-instance-attribute` | `--instance-id`, `--attribute` (instanceType, userData, instanceInitiatedShutdownBehavior, disableApiTermination, disableApiStop, ebsOptimized, enaSupport, sourceDestCheck, rootDeviceName, kernel, ramdisk) | `--dry-run` | Instance must exist (running or stopped) | Gateway validates input → NATS `ec2.DescribeInstanceAttribute` with `spinifex-workers` queue group → daemon checks running instances first (`d.Instances.VMS`), then stopped instances in JetStream KV → returns single attribute per call (matches AWS behavior). Stored attributes (`instanceType`, `userData`) return real values; unstored attributes return AWS defaults (`instanceInitiatedShutdownBehavior`=stop, `disableApiTermination`=false, etc.) | 1. Get instanceType from running instance<br>2. Get userData from stopped instance<br>3. Get default disableApiTermination<br>4. Invalid attribute name (error)<br>5. Instance not found (error: InvalidInstanceID.NotFound) | **DONE** |
| `describe-instance-attribute` | `--instance-id`, `--attribute` (instanceType, userData, instanceInitiatedShutdownBehavior, disableApiTermination, disableApiStop, ebsOptimized, enaSupport, sourceDestCheck, rootDeviceName, kernel, ramdisk) | `--dry-run` | Instance must exist (running or stopped) | Gateway validates input → NATS `ec2.DescribeInstanceAttribute` with `spinifex-workers` queue group → daemon checks running instances first (`d.Instances.VMS`), then stopped instances in JetStream KV → returns single attribute per call (matches AWS behavior). Stored attributes (`instanceType`, `userData`, `disableApiTermination`) return real values from `RunInstancesInput`; remaining attributes return AWS defaults (`instanceInitiatedShutdownBehavior`=stop, etc.) | 1. Get instanceType from running instance<br>2. Get userData from stopped instance<br>3. Get disableApiTermination (reflects RunInstances flag + Modify updates)<br>4. Invalid attribute name (error)<br>5. Instance not found (error: InvalidInstanceID.NotFound) | **DONE** |
| `describe-instance-credit-specifications` | `--instance-ids` | `--filters`, `--max-results`, `--dry-run` | None | Gateway-only stub — returns `CpuCredits: "standard"` for each requested instance ID. No daemon round-trip. T-series credit mode is not persisted. | 1. Get credit spec for T-series instance<br>2. Multiple instance IDs | **DONE** |
| `describe-instance-status` | `--instance-ids`, `--include-all-instances`, `--filters` (availability-zone, instance-state-code, instance-state-name, tag:\*) | `--max-results`, `--next-token`, `--dry-run` (silently ignored), `event.*` / `instance-status.*` / `system-status.*` filters (rejected with InvalidParameterValue) | None | Gateway fans out NATS `ec2.DescribeInstanceStatus` to all nodes (no queue group). Each daemon validates input (i- prefix, filter names), iterates local VMs and renders InstanceStatus: `Running`→ok/passed, everything-else-included→not-applicable. When `IncludeAllInstances=true`, gateway also queries `ec2.DescribeStoppedInstances` in parallel and transforms reservation entries into InstanceStatus with `not-applicable`. Aggregated results dedup'd by InstanceId (running > stopped). Terminated and error-state instances never returned. No real health probing — static values reflect Mulga's lifecycle state only. | 1. Status of running instance (ok/passed)<br>2. Filter by instance ID<br>3. Default excludes stopped instances<br>4. IncludeAllInstances=true includes stopped (not-applicable)<br>5. Cross-account instance excluded<br>6. Unknown filter returns InvalidParameterValue<br>7. Malformed instance ID rejected<br>8. Tag filter narrows correctly | **DONE** |
| `monitor-instances` | — | `--instance-ids` | Instance must exist | Enable basic monitoring (CPU, disk, network) → store metrics in NATS KV → return monitoring state | 1. Enable monitoring<br>2. Verify monitoring state in describe-instances | **NOT STARTED** |
+1 -109
View File
@@ -443,113 +443,5 @@ func (d *Daemon) handleEC2ModifyInstanceAttribute(msg *nats.Msg) {
// handleEC2DescribeInstanceAttribute returns a single requested attribute for an instance.
func (d *Daemon) handleEC2DescribeInstanceAttribute(msg *nats.Msg) {
var input ec2.DescribeInstanceAttributeInput
if err := json.Unmarshal(msg.Data, &input); err != nil {
slog.Error("handleEC2DescribeInstanceAttribute: failed to unmarshal request", "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
if input.InstanceId == nil || *input.InstanceId == "" {
slog.Error("handleEC2DescribeInstanceAttribute: missing instance_id")
respondWithError(msg, awserrors.ErrorMissingParameter)
return
}
if input.Attribute == nil || *input.Attribute == "" {
slog.Error("handleEC2DescribeInstanceAttribute: missing attribute")
respondWithError(msg, awserrors.ErrorMissingParameter)
return
}
instanceID := *input.InstanceId
attribute := *input.Attribute
accountID := utils.AccountIDFromMsg(msg)
// Look up instance: running first, then stopped KV.
var instance *vm.VM
if running, ok := d.vmMgr.Get(instanceID); ok {
instance = running
}
if instance == nil {
if d.stateStore == nil {
slog.Error("handleEC2DescribeInstanceAttribute: state store not available")
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
stopped, err := d.stateStore.LoadStoppedInstance(instanceID)
if err != nil {
slog.Error("handleEC2DescribeInstanceAttribute: failed to load stopped instance",
"instanceId", instanceID, "err", err)
respondWithError(msg, awserrors.ErrorServerInternal)
return
}
instance = stopped
}
if instance == nil {
slog.Warn("handleEC2DescribeInstanceAttribute: instance not found",
"instanceId", instanceID)
respondWithError(msg, awserrors.ErrorInvalidInstanceIDNotFound)
return
}
if !checkInstanceOwnership(msg, instanceID, instance.AccountID) {
return
}
output := &ec2.DescribeInstanceAttributeOutput{
InstanceId: &instanceID,
}
switch attribute {
case ec2.InstanceAttributeNameInstanceType:
val := instance.InstanceType
output.InstanceType = &ec2.AttributeValue{Value: &val}
case ec2.InstanceAttributeNameUserData:
val := instance.UserData
output.UserData = &ec2.AttributeValue{Value: &val}
case ec2.InstanceAttributeNameDisableApiTermination:
val := false
output.DisableApiTermination = &ec2.AttributeBooleanValue{Value: &val}
case ec2.InstanceAttributeNameDisableApiStop:
val := false
output.DisableApiStop = &ec2.AttributeBooleanValue{Value: &val}
case ec2.InstanceAttributeNameInstanceInitiatedShutdownBehavior:
val := ec2.ShutdownBehaviorStop
output.InstanceInitiatedShutdownBehavior = &ec2.AttributeValue{Value: &val}
case ec2.InstanceAttributeNameEbsOptimized:
val := false
output.EbsOptimized = &ec2.AttributeBooleanValue{Value: &val}
case ec2.InstanceAttributeNameEnaSupport:
val := true
output.EnaSupport = &ec2.AttributeBooleanValue{Value: &val}
case ec2.InstanceAttributeNameSourceDestCheck:
val := true
output.SourceDestCheck = &ec2.AttributeBooleanValue{Value: &val}
case ec2.InstanceAttributeNameGroupSet:
if instance.Instance != nil && len(instance.Instance.SecurityGroups) > 0 {
output.Groups = instance.Instance.SecurityGroups
} else {
output.Groups = []*ec2.GroupIdentifier{}
}
default:
slog.Warn("handleEC2DescribeInstanceAttribute: unsupported attribute",
"instanceId", instanceID, "attribute", attribute)
respondWithError(msg, awserrors.ErrorInvalidParameterValue)
return
}
slog.Info("handleEC2DescribeInstanceAttribute: completed",
"instanceId", instanceID, "attribute", attribute, "accountID", accountID)
respondWithJSON(msg, output)
handleNATSRequest(msg, d.instanceService.DescribeInstanceAttribute)
}
+9 -2
View File
@@ -1691,6 +1691,9 @@ func TestHandleEC2DescribeInstanceAttribute_DefaultAttribute_DisableApiTerminati
require.NoError(t, err)
defer sub.Unsubscribe()
// Smoke test the daemon→service delegate: write a stopped instance with
// the protection flag set and confirm the round-trip surfaces the real
// value (not the pre-migration hardcoded false).
instanceID := "i-describe-def-001"
instance := &vm.VM{
ID: instanceID,
@@ -1700,6 +1703,9 @@ func TestHandleEC2DescribeInstanceAttribute_DefaultAttribute_DisableApiTerminati
Instance: &ec2.Instance{
InstanceId: aws.String(instanceID),
},
RunInstancesInput: &ec2.RunInstancesInput{
DisableApiTermination: aws.Bool(true),
},
}
err = daemon.jsManager.WriteStoppedInstance(instanceID, instance)
require.NoError(t, err)
@@ -1717,7 +1723,7 @@ func TestHandleEC2DescribeInstanceAttribute_DefaultAttribute_DisableApiTerminati
err = json.Unmarshal(reply.Data, &output)
require.NoError(t, err)
require.NotNil(t, output.DisableApiTermination)
assert.Equal(t, false, *output.DisableApiTermination.Value)
assert.Equal(t, true, *output.DisableApiTermination.Value)
}
func TestHandleEC2DescribeInstanceAttribute_DefaultAttribute_ShutdownBehavior(t *testing.T) {
@@ -1915,7 +1921,8 @@ func TestHandleEC2DescribeInstanceAttribute_InvalidJSON(t *testing.T) {
var errResp map[string]any
err = json.Unmarshal(reply.Data, &errResp)
require.NoError(t, err)
assert.Equal(t, awserrors.ErrorServerInternal, errResp["Code"])
// utils.UnmarshalJsonPayload returns ValidationError on parse failure.
assert.Equal(t, awserrors.ErrorValidationError, errResp["Code"])
}
// --- Delegate handler round-trip tests (table-driven) ---
@@ -38,6 +38,9 @@ func ValidateModifyInstanceAttributeInput(input *ec2.ModifyInstanceAttributeInpu
if input.SourceDestCheck != nil {
count++
}
if input.DisableApiTermination != nil {
count++
}
if count != 1 {
return errors.New(awserrors.ErrorInvalidParameterValue)
}
@@ -95,6 +95,24 @@ func TestValidateModifyInstanceAttributeInput_ValidUserData(t *testing.T) {
assert.NoError(t, err)
}
func TestValidateModifyInstanceAttributeInput_ValidDisableApiTermination(t *testing.T) {
err := ValidateModifyInstanceAttributeInput(&ec2.ModifyInstanceAttributeInput{
InstanceId: aws.String("i-abc123"),
DisableApiTermination: &ec2.AttributeBooleanValue{Value: aws.Bool(true)},
})
assert.NoError(t, err)
}
func TestValidateModifyInstanceAttributeInput_DisableApiTerminationWithOther(t *testing.T) {
err := ValidateModifyInstanceAttributeInput(&ec2.ModifyInstanceAttributeInput{
InstanceId: aws.String("i-abc123"),
DisableApiTermination: &ec2.AttributeBooleanValue{Value: aws.Bool(true)},
InstanceType: &ec2.AttributeValue{Value: aws.String("t3.micro")},
})
require.Error(t, err)
assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error())
}
func TestValidateModifyInstanceAttributeInput_ValidSourceDestCheck(t *testing.T) {
err := ValidateModifyInstanceAttributeInput(&ec2.ModifyInstanceAttributeInput{
InstanceId: aws.String("i-abc123"),
+103 -21
View File
@@ -731,29 +731,54 @@ func (s *InstanceServiceImpl) StopOrTerminateInstance(instance *vm.VM, command s
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 {
// Fold the termination-protection check, idempotency check, transition
// validation, and attribute stamp into one lock acquisition so a racing
// ModifyInstanceAttribute can't clear protection between the gates. The
// async lifecycle goroutine below persists on its own state transitions,
// so no synchronous persist is needed here.
var (
protected, raced, stateMismatch bool
currentState vm.InstanceState
)
ok := s.vmMgr.UpdateState(instance.ID, func(v *vm.VM) {
currentState = v.Status
if isTerminate && v.IsTerminationProtected() {
protected = true
return
}
if isTerminate && v.Status == vm.StateShuttingDown {
raced = true
return
}
if !vm.IsValidTransition(v.Status, initialState) {
stateMismatch = true
return
}
v.Attributes = command.Attributes
})
if !ok {
slog.Warn("StopOrTerminateInstance: instance no longer in running map",
"instanceId", instance.ID)
return errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
}
if protected {
slog.Warn("StopOrTerminateInstance: instance has termination protection",
"instanceId", instance.ID)
return errors.New(awserrors.ErrorOperationNotPermitted)
}
if raced {
// Idempotent: a concurrent terminate goroutine is already cleaning up.
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) {
if stateMismatch {
// Surface IncorrectInstanceState synchronously so the AWS SDK sees
// 400 instead of a stale 200. The async path re-validates.
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 {
@@ -1641,11 +1666,11 @@ 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.
// ModifyInstanceAttribute applies a single attribute change. InstanceType and
// UserData require the instance to be stopped. DisableApiTermination works on
// both running and stopped instances. 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)
@@ -1657,6 +1682,41 @@ func (s *InstanceServiceImpl) ModifyInstanceAttribute(input *ec2.ModifyInstanceA
return &ec2.ModifyInstanceAttributeOutput{}, nil
}
// DisableApiTermination applies to both running and stopped instances.
// Try the running map first; fall through to the stopped-store path only
// if the instance isn't currently running on this node.
if input.DisableApiTermination != nil && input.DisableApiTermination.Value != nil {
newVal := input.DisableApiTermination.Value
var notVisible bool
updated, persistErr := s.vmMgr.UpdateAndPersist(instanceID, func(v *vm.VM) bool {
if !IsInstanceVisible(accountID, v.AccountID) {
notVisible = true
return false
}
if v.RunInstancesInput == nil {
v.RunInstancesInput = &ec2.RunInstancesInput{}
}
v.RunInstancesInput.DisableApiTermination = newVal
return true
})
if notVisible {
slog.Warn("ModifyInstanceAttribute: instance not visible",
"instanceId", instanceID, "callerAccount", accountID)
return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
}
if updated {
if persistErr != nil {
slog.Error("ModifyInstanceAttribute: persist failed",
"instanceId", instanceID, "err", persistErr)
return nil, errors.New(awserrors.ErrorServerInternal)
}
slog.Info("ModifyInstanceAttribute: updated DisableApiTermination on running instance",
"instanceId", instanceID, "value", *newVal)
return &ec2.ModifyInstanceAttributeOutput{}, nil
}
// Not in running map — fall through to stopped-store path.
}
if s.stoppedStore == nil {
slog.Error("ModifyInstanceAttribute: stopped store not available")
return nil, errors.New(awserrors.ErrorServerInternal)
@@ -1711,6 +1771,15 @@ func (s *InstanceServiceImpl) ModifyInstanceAttribute(input *ec2.ModifyInstanceA
}
}
if input.DisableApiTermination != nil && input.DisableApiTermination.Value != nil {
slog.Info("ModifyInstanceAttribute: changing DisableApiTermination on stopped instance",
"instanceId", instanceID, "value", *input.DisableApiTermination.Value)
if instance.RunInstancesInput == nil {
instance.RunInstancesInput = &ec2.RunInstancesInput{}
}
instance.RunInstancesInput.DisableApiTermination = input.DisableApiTermination.Value
}
if err := s.stoppedStore.WriteStoppedInstance(instanceID, instance); err != nil {
slog.Error("ModifyInstanceAttribute: failed to write modified instance to KV",
"instanceId", instanceID, "err", err)
@@ -1856,6 +1925,12 @@ func (s *InstanceServiceImpl) TerminateStoppedInstance(input *TerminateStoppedIn
return nil, errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
}
if instance.IsTerminationProtected() {
slog.Warn("TerminateStoppedInstance: instance has termination protection",
"instanceId", input.InstanceID)
return nil, errors.New(awserrors.ErrorOperationNotPermitted)
}
s.deleteInstanceVolumes(instance, input.InstanceID)
s.releaseInstancePublicIP(instance, input.InstanceID)
s.deleteInstanceENI(instance, input.InstanceID)
@@ -2018,7 +2093,14 @@ func (s *InstanceServiceImpl) DescribeInstanceAttribute(input *ec2.DescribeInsta
output.UserData = &ec2.AttributeValue{Value: &val}
case ec2.InstanceAttributeNameDisableApiTermination:
val := false
// Read under the manager lock so we serialise with a concurrent
// ModifyInstanceAttribute writer. Inspect is a no-op race-wise for
// stopped instances (no concurrent writers) but keeps the call site
// uniform.
var val bool
s.vmMgr.Inspect(instance, func(v *vm.VM) {
val = v.IsTerminationProtected()
})
output.DisableApiTermination = &ec2.AttributeBooleanValue{Value: &val}
case ec2.InstanceAttributeNameDisableApiStop:
@@ -987,6 +987,63 @@ func TestDescribeInstanceAttribute_HiddenForOtherAccount(t *testing.T) {
assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
}
func TestDescribeInstanceAttribute_DisableApiTermination(t *testing.T) {
owner := utils.GlobalAccountID
tests := []struct {
name string
instance *vm.VM
want bool
}{
{
name: "flag true on RunInstancesInput",
instance: &vm.VM{
ID: "i-prot", AccountID: owner,
RunInstancesInput: &ec2.RunInstancesInput{
DisableApiTermination: aws.Bool(true),
},
},
want: true,
},
{
name: "flag false on RunInstancesInput",
instance: &vm.VM{
ID: "i-noprot", AccountID: owner,
RunInstancesInput: &ec2.RunInstancesInput{
DisableApiTermination: aws.Bool(false),
},
},
want: false,
},
{
name: "RunInstancesInput.DisableApiTermination nil",
instance: &vm.VM{
ID: "i-nilflag", AccountID: owner,
RunInstancesInput: &ec2.RunInstancesInput{},
},
want: false,
},
{
name: "RunInstancesInput nil (legacy)",
instance: &vm.VM{ID: "i-legacy", AccountID: owner},
want: false,
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
svc := &InstanceServiceImpl{vmMgr: mgrWith(map[string]*vm.VM{tc.instance.ID: tc.instance})}
out, err := svc.DescribeInstanceAttribute(&ec2.DescribeInstanceAttributeInput{
InstanceId: aws.String(tc.instance.ID),
Attribute: aws.String(ec2.InstanceAttributeNameDisableApiTermination),
}, owner)
require.NoError(t, err)
require.NotNil(t, out.DisableApiTermination)
assert.Equal(t, tc.want, *out.DisableApiTermination.Value)
})
}
}
func TestDescribeStoppedInstances_NilStore(t *testing.T) {
svc := &InstanceServiceImpl{}
_, err := svc.DescribeStoppedInstances(&ec2.DescribeInstancesInput{}, "")
@@ -1257,6 +1314,81 @@ func TestModifyInstanceAttribute_ClearsStateReason(t *testing.T) {
assert.Nil(t, updated.Instance.StateReason)
}
func TestModifyInstanceAttribute_DisableApiTermination_Running(t *testing.T) {
const owner = "acc"
tests := []struct {
name string
initial *ec2.RunInstancesInput
setTo bool
wantFlag bool
}{
{name: "set true on empty input", initial: &ec2.RunInstancesInput{}, setTo: true, wantFlag: true},
{name: "set true on nil input (legacy)", initial: nil, setTo: true, wantFlag: true},
{name: "clear from true to false", initial: &ec2.RunInstancesInput{DisableApiTermination: aws.Bool(true)}, setTo: false, wantFlag: false},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
id := "i-run"
mgr := mgrWith(map[string]*vm.VM{
id: {ID: id, AccountID: owner, Status: vm.StateRunning, RunInstancesInput: tc.initial},
})
svc := &InstanceServiceImpl{vmMgr: mgr}
_, err := svc.ModifyInstanceAttribute(&ec2.ModifyInstanceAttributeInput{
InstanceId: aws.String(id),
DisableApiTermination: &ec2.AttributeBooleanValue{Value: aws.Bool(tc.setTo)},
}, owner)
require.NoError(t, err)
got, _ := mgr.Get(id)
require.NotNil(t, got.RunInstancesInput)
require.NotNil(t, got.RunInstancesInput.DisableApiTermination)
assert.Equal(t, tc.wantFlag, *got.RunInstancesInput.DisableApiTermination)
})
}
}
func TestModifyInstanceAttribute_DisableApiTermination_Running_NotVisible(t *testing.T) {
id := "i-run-other"
mgr := mgrWith(map[string]*vm.VM{
id: {ID: id, AccountID: "owner-acc", Status: vm.StateRunning},
})
svc := &InstanceServiceImpl{vmMgr: mgr}
_, err := svc.ModifyInstanceAttribute(&ec2.ModifyInstanceAttributeInput{
InstanceId: aws.String(id),
DisableApiTermination: &ec2.AttributeBooleanValue{Value: aws.Bool(true)},
}, "other-acc")
require.Error(t, err)
assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
}
func TestModifyInstanceAttribute_DisableApiTermination_Stopped(t *testing.T) {
id := "i-stop-prot"
owner := "acc"
store := &fakeStoppedStore{loadByID: map[string]*vm.VM{
id: {
ID: id, Status: vm.StateStopped, AccountID: owner,
Instance: &ec2.Instance{InstanceId: aws.String(id)},
},
}}
svc := &InstanceServiceImpl{
vmMgr: mgrWith(map[string]*vm.VM{}),
stoppedStore: store,
}
_, err := svc.ModifyInstanceAttribute(&ec2.ModifyInstanceAttributeInput{
InstanceId: aws.String(id),
DisableApiTermination: &ec2.AttributeBooleanValue{Value: aws.Bool(true)},
}, owner)
require.NoError(t, err)
updated := store.wroteStopped[id]
require.NotNil(t, updated)
require.NotNil(t, updated.RunInstancesInput)
assert.True(t, *updated.RunInstancesInput.DisableApiTermination)
}
func TestModifyInstanceAttribute_WriteError(t *testing.T) {
id := "i-werr"
store := &fakeStoppedStore{
@@ -1407,6 +1539,30 @@ func TestTerminateStoppedInstance_NotVisible(t *testing.T) {
assert.Equal(t, awserrors.ErrorInvalidInstanceIDNotFound, err.Error())
}
func TestTerminateStoppedInstance_TerminationProtected(t *testing.T) {
id := "i-prot"
v := &vm.VM{
ID: id, Status: vm.StateStopped, AccountID: "acc",
RunInstancesInput: &ec2.RunInstancesInput{
DisableApiTermination: aws.Bool(true),
},
}
v.EBSRequests.Requests = []spxtypes.EBSRequest{
{Name: "vol-001", DeleteOnTermination: true},
}
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.Error(t, err)
assert.Equal(t, awserrors.ErrorOperationNotPermitted, err.Error())
assert.Empty(t, vd.calls, "volumes must not be deleted when termination protected")
assert.Empty(t, store.wroteTerminated, "must not write to terminated bucket when protected")
assert.Empty(t, store.deletedStopped, "must not remove from stopped bucket when protected")
}
func TestTerminateStoppedInstance_HappyPath(t *testing.T) {
id := "i-term-001"
store := &fakeStoppedStore{loadByID: map[string]*vm.VM{
@@ -1946,6 +2102,40 @@ func TestStopOrTerminateInstance_InvalidTransition(t *testing.T) {
assert.Equal(t, awserrors.ErrorIncorrectInstanceState, err.Error())
}
func TestStopOrTerminateInstance_TerminationProtection(t *testing.T) {
// DisableApiTermination must block Terminate but never Stop.
tests := []struct {
name string
attrs spxtypes.EC2CommandAttributes
wantErr string
}{
{name: "terminate blocked", attrs: spxtypes.EC2CommandAttributes{TerminateInstance: true}, wantErr: awserrors.ErrorOperationNotPermitted},
{name: "stop allowed", attrs: spxtypes.EC2CommandAttributes{StopInstance: true}, wantErr: ""},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
id := "i-prot"
mgr := mgrWith(map[string]*vm.VM{
id: {
ID: id, Status: vm.StateRunning,
RunInstancesInput: &ec2.RunInstancesInput{DisableApiTermination: aws.Bool(true)},
},
})
v, _ := mgr.Get(id)
svc := &InstanceServiceImpl{vmMgr: mgr}
err := svc.StopOrTerminateInstance(v, spxtypes.EC2InstanceCommand{ID: id, Attributes: tc.attrs})
if tc.wantErr == "" {
require.NoError(t, err)
return
}
require.Error(t, err)
assert.Equal(t, tc.wantErr, err.Error())
assert.Equal(t, vm.StateRunning, v.Status, "VM state must not change when blocked")
})
}
}
type fakeENICreator struct {
defaultSubnet *SubnetInfo
subnet *SubnetInfo
+19
View File
@@ -219,6 +219,25 @@ func (m *Manager) UpdateState(id string, fn func(*VM)) bool {
return true
}
// UpdateAndPersist looks up id and runs fn(v) under the manager lock. fn
// returns true to request a state-store persist after the lock is released
// (matching the writeRunningState pattern used by AttachVolume / Stop /
// Terminate); false signals a no-op or early-return mutator and skips the
// persist entirely. Returns (false, nil) when the VM is not in the running
// map; fn is not invoked. On persist failure the in-memory mutation has
// already taken effect — the caller's API-level error masks the drift but
// memory will be ahead of disk until the next successful persist.
func (m *Manager) UpdateAndPersist(id string, fn func(*VM) bool) (bool, error) {
var changed bool
found := m.UpdateState(id, func(v *VM) {
changed = fn(v)
})
if !found || !changed {
return found, nil
}
return true, m.writeRunningState()
}
// Status returns v.Status under the manager lock. Replaces the dominant
// "Inspect to read Status" pattern with a typed accessor. Read-only — no
// membership check, so callers may pass a pointer to an instance no longer
+4
View File
@@ -12,6 +12,7 @@ type fakeStateStore struct {
saved map[string]map[string]*VM
stopped map[string]*VM
terminated map[string]*VM
saveErr error
}
func newFakeStateStore() *fakeStateStore {
@@ -23,6 +24,9 @@ func newFakeStateStore() *fakeStateStore {
}
func (f *fakeStateStore) SaveRunningState(nodeID string, snap map[string]*VM) error {
if f.saveErr != nil {
return f.saveErr
}
cp := make(map[string]*VM, len(snap))
maps.Copy(cp, snap)
f.saved[nodeID] = cp
+115
View File
@@ -1,6 +1,7 @@
package vm
import (
"errors"
"sync"
"sync/atomic"
"testing"
@@ -98,6 +99,120 @@ func TestManager_UpdateState(t *testing.T) {
}
}
func TestManager_UpdateAndPersist(t *testing.T) {
t.Run("persists on hit when mutator reports change", func(t *testing.T) {
store := newFakeStateStore()
m := NewManagerWithDeps(Deps{NodeID: "node-1", StateStore: store})
a := mkVM("i-a")
a.Status = StatePending
m.Insert(a)
called := false
ok, err := m.UpdateAndPersist("i-a", func(v *VM) bool {
called = true
v.Status = StateRunning
return true
})
if !ok {
t.Fatal("UpdateAndPersist on existing id: want ok=true")
}
if err != nil {
t.Fatalf("UpdateAndPersist err: %v", err)
}
if !called {
t.Fatal("mutator not invoked")
}
if a.Status != StateRunning {
t.Fatalf("mutation not applied: got %s, want %s", a.Status, StateRunning)
}
saved, ok := store.saved["node-1"]
if !ok {
t.Fatal("StateStore.SaveRunningState not called for node-1")
}
if _, ok := saved["i-a"]; !ok {
t.Fatal("persisted snapshot missing i-a")
}
})
t.Run("mutator returning false skips persist", func(t *testing.T) {
store := newFakeStateStore()
m := NewManagerWithDeps(Deps{NodeID: "node-1", StateStore: store})
a := mkVM("i-a")
m.Insert(a)
ok, err := m.UpdateAndPersist("i-a", func(v *VM) bool { return false })
if !ok {
t.Fatal("UpdateAndPersist on existing id: want ok=true")
}
if err != nil {
t.Fatalf("UpdateAndPersist err: %v", err)
}
if len(store.saved) != 0 {
t.Fatal("SaveRunningState must not be called when mutator reports no change")
}
})
t.Run("missing id is no-op", func(t *testing.T) {
store := newFakeStateStore()
m := NewManagerWithDeps(Deps{NodeID: "node-1", StateStore: store})
ok, err := m.UpdateAndPersist("missing", func(*VM) bool {
t.Fatal("mutator must not run on missing id")
return false
})
if ok {
t.Fatal("UpdateAndPersist on missing id: want ok=false")
}
if err != nil {
t.Fatalf("UpdateAndPersist err: %v", err)
}
if len(store.saved) != 0 {
t.Fatal("SaveRunningState must not be called on miss")
}
})
t.Run("nil state store still mutates", func(t *testing.T) {
m := NewManager()
a := mkVM("i-a")
m.Insert(a)
ok, err := m.UpdateAndPersist("i-a", func(v *VM) bool {
v.Status = StateRunning
return true
})
if !ok || err != nil {
t.Fatalf("UpdateAndPersist: ok=%v err=%v, want (true, nil)", ok, err)
}
if a.Status != StateRunning {
t.Fatalf("mutation not applied: got %s", a.Status)
}
})
t.Run("persist error is surfaced; in-memory mutation retained", func(t *testing.T) {
store := newFakeStateStore()
want := errors.New("save failed")
store.saveErr = want
m := NewManagerWithDeps(Deps{NodeID: "node-1", StateStore: store})
a := mkVM("i-a")
a.Status = StatePending
m.Insert(a)
ok, err := m.UpdateAndPersist("i-a", func(v *VM) bool {
v.Status = StateRunning
return true
})
if !ok {
t.Fatal("UpdateAndPersist on existing id: want ok=true")
}
if !errors.Is(err, want) {
t.Fatalf("UpdateAndPersist err: got %v, want %v", err, want)
}
if a.Status != StateRunning {
t.Fatalf("in-memory mutation must persist past persist failure: got %s", a.Status)
}
})
}
func TestManager_Inspect(t *testing.T) {
m := NewManager()
a := mkVM("i-a")
+12
View File
@@ -10,6 +10,7 @@ import (
"sync"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/qmp"
"github.com/mulgadc/spinifex/spinifex/types"
@@ -136,6 +137,17 @@ func (v *VM) ResetNodeLocalState() {
v.EBSRequests.Mu = sync.Mutex{}
}
// IsTerminationProtected reports whether the instance was launched with
// DisableApiTermination=true and that value has not been cleared via
// ModifyInstanceAttribute. Nil-safe for legacy instances missing
// RunInstancesInput.
func (v *VM) IsTerminationProtected() bool {
if v == nil || v.RunInstancesInput == nil {
return false
}
return aws.BoolValue(v.RunInstancesInput.DisableApiTermination)
}
type NetDev struct {
Value string `json:"value"`
}