diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index 2d6134f4e..363306f2a 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -113,7 +113,7 @@ Operational commands for inspecting cluster state. These fan out NATS requests t | Command | Implemented Flags | Missing Flags | Status | |---------|-------------------|---------------|--------| -| `run-instances` | `--image-id`, `--instance-type`, `--count`, `--key-name`, `--user-data`, `--subnet-id`, `--security-group-ids`, `--tag-specifications` (instance-scoped), `--block-device-mappings` (DeviceName, VolumeSize, VolumeType, Iops, DeleteOnTermination), `--placement` (GroupName), `--iam-instance-profile` (Name/Arn) | `--dry-run`, `--client-token`, `--disable-api-termination`, `--ebs-optimized`, `--network-interfaces`, `--private-ip-address`, `--monitoring`, `--credit-specification`, `--cpu-options`, `--metadata-options`, `--launch-template`, `--hibernate-options` | **DONE** | +| `run-instances` | `--image-id`, `--instance-type`, `--count`, `--key-name`, `--user-data`, `--subnet-id`, `--security-group-ids`, `--tag-specifications` (instance-scoped), `--block-device-mappings` (DeviceName, VolumeSize, VolumeType, Iops, DeleteOnTermination), `--placement` (GroupName), `--iam-instance-profile` (Name/Arn), `--capacity-reservation-specification` (CapacityReservationTarget.CapacityReservationId, targeted-by-id only) | `--dry-run`, `--client-token`, `--disable-api-termination`, `--ebs-optimized`, `--network-interfaces`, `--private-ip-address`, `--monitoring`, `--credit-specification`, `--cpu-options`, `--metadata-options`, `--launch-template`, `--hibernate-options` | **DONE** | | `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` | **DONE** | | `start-instances` | `--instance-ids` | `--dry-run`, `--force` | **DONE** | | `stop-instances` | `--instance-ids` | `--force`, `--hibernate`, `--dry-run` | **DONE** | @@ -132,6 +132,11 @@ Operational commands for inspecting cluster state. These fan out NATS requests t role ARN and rejects cross-account profile ARNs. Placement strategies: `spread` (1 per node, atomic CAS) and `cluster` (pin all to one node). +`run-instances --capacity-reservation-specification` consumes a reserved slot on +the node owning the targeted reservation (targeted-by-id only; `open` +auto-matching is not supported). The combination with `--placement` (group) is +rejected. `describe-instances` then reports the instance's `CapacityReservationId`. + ### EC2 — IAM Instance Profile Associations Stored as `vm.VM.IamInstanceProfileArn` + `IamInstanceProfileAssociationId` diff --git a/spinifex/daemon/capacityreservation.go b/spinifex/daemon/capacityreservation.go index bc3131a1e..720ba47ab 100644 --- a/spinifex/daemon/capacityreservation.go +++ b/spinifex/daemon/capacityreservation.go @@ -4,6 +4,8 @@ import ( "errors" "time" + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/awserrors" ) @@ -22,10 +24,13 @@ type capacityReservation struct { Tenancy string InstancePlatform string CreateDate time.Time + ConsumedCount int // instances launched into this reservation; slots moved reservedCR*->allocated* } -// reservationCompute returns the headline compute carve-out for a reservation: -// InstanceCount × catalog (vCPU, memory), with no nbdkit overhead. +// reservationCompute returns the full compute carve-out for a reservation: +// TotalInstanceCount × per-instance (vCPU, memory). MemGBPerInstance already +// folds in nbdkit overhead (set at create), so the reservedCR->allocated swap on +// launch is net-zero. func (r *capacityReservation) reservationCompute() (vcpu int, memGB float64) { return r.TotalInstanceCount * r.VCPUPerInstance, float64(r.TotalInstanceCount) * r.MemGBPerInstance } @@ -52,10 +57,10 @@ func (rm *ResourceManager) CreateReservation(rec *capacityReservation) error { return nil } -// CancelReservation releases the carve-out for an account-owned reservation and -// removes it from the map, then refreshes subscriptions. Returns the removed -// record and true on success, or (nil, false) if no reservation with that id is -// owned by the account. +// CancelReservation releases the unconsumed remainder of an account-owned +// reservation's carve-out and removes it; consumed slots already moved to +// allocated* and keep running. Returns the removed record and true on success, +// or (nil, false) if no reservation with that id is owned by the account. func (rm *ResourceManager) CancelReservation(id, accountID string) (*capacityReservation, bool) { rm.mu.Lock() rec, ok := rm.reservations[id] @@ -63,9 +68,9 @@ func (rm *ResourceManager) CancelReservation(id, accountID string) (*capacityRes rm.mu.Unlock() return nil, false } - freedVCPU, freedMem := rec.reservationCompute() - rm.reservedCRVCPU -= freedVCPU - rm.reservedCRMem -= freedMem + remaining := rec.TotalInstanceCount - rec.ConsumedCount + rm.reservedCRVCPU -= remaining * rec.VCPUPerInstance + rm.reservedCRMem -= float64(remaining) * rec.MemGBPerInstance delete(rm.reservations, id) rm.mu.Unlock() @@ -73,16 +78,105 @@ func (rm *ResourceManager) CancelReservation(id, accountID string) (*capacityRes return rec, true } -// ListReservations returns a snapshot of the account's reservations on this node. -// Records are immutable, so the returned pointers are safe to read concurrently. -func (rm *ResourceManager) ListReservations(accountID string) []*capacityReservation { +// AllocateFromReservation launches one instance into reservation crID via a +// net-zero swap under rm.mu: the per-instance compute already held in reservedCR* +// at create moves to allocated* and ConsumedCount bumps, so schedulable capacity +// is unchanged by construction. Errors: unknown/foreign id -> +// InvalidCapacityReservationId.NotFound; type mismatch -> InvalidParameterValue; +// already full -> ReservationCapacityExceeded. +func (rm *ResourceManager) AllocateFromReservation(crID, accountID string, it *ec2.InstanceTypeInfo) error { + rm.mu.Lock() + rec, ok := rm.reservations[crID] + if !ok || rec.AccountID != accountID { + rm.mu.Unlock() + return errors.New(awserrors.ErrorInvalidCapacityReservationIdNotFound) + } + if rec.InstanceType != aws.StringValue(it.InstanceType) { + rm.mu.Unlock() + return errors.New(awserrors.ErrorInvalidParameterValue) + } + if rec.ConsumedCount >= rec.TotalInstanceCount { + rm.mu.Unlock() + return errors.New(awserrors.ErrorReservationCapacityExceeded) + } + rm.reservedCRVCPU -= rec.VCPUPerInstance + rm.reservedCRMem -= rec.MemGBPerInstance + rm.allocatedVCPU += rec.VCPUPerInstance + rm.allocatedMem += rec.MemGBPerInstance + rec.ConsumedCount++ + rm.mu.Unlock() + + rm.updateInstanceSubscriptions() + return nil +} + +// ReleaseToReservation returns a launched instance's slot. If the reservation +// still exists it is the inverse net-zero swap (allocated* -> reservedCR*, +// ConsumedCount--); if the reservation is gone (cancelled or lost on restart) the +// allocation is freed to the general pool using the catalog charge so it never +// overcounts. Best-effort: a missing reservation is not an error. +func (rm *ResourceManager) ReleaseToReservation(crID string, it *ec2.InstanceTypeInfo) { + rm.mu.Lock() + if rec, ok := rm.reservations[crID]; ok && rec.ConsumedCount > 0 { + rm.allocatedVCPU -= rec.VCPUPerInstance + rm.allocatedMem -= rec.MemGBPerInstance + rm.reservedCRVCPU += rec.VCPUPerInstance + rm.reservedCRMem += rec.MemGBPerInstance + rec.ConsumedCount-- + } else { + rm.allocatedVCPU -= int(instanceTypeVCPUs(it)) + rm.allocatedMem -= float64(rm.instanceMemChargeMiB(it)) / 1024.0 + } + rm.mu.Unlock() + + rm.updateInstanceSubscriptions() +} + +// ReservationAvailable reports how many more instances can launch into crID +// (Total - Consumed), or 0 when the reservation is unknown, owned by another +// account, or for a different instance type. The daemon handler's up-front check +// turns those zero cases into precise errors before the launch loop. +func (rm *ResourceManager) ReservationAvailable(crID, accountID string, it *ec2.InstanceTypeInfo) int { + rm.mu.RLock() + defer rm.mu.RUnlock() + rec, ok := rm.reservations[crID] + if !ok || rec.AccountID != accountID || rec.InstanceType != aws.StringValue(it.InstanceType) { + return 0 + } + return rec.TotalInstanceCount - rec.ConsumedCount +} + +// ValidateReservationTarget is the owning daemon's up-front semantic check for a +// targeted launch, run before the launch loop. It returns +// InvalidCapacityReservationId.NotFound for an unknown or foreign id and +// InvalidParameterValue on an instance-type mismatch. The "full" case is left to +// the count gate in PrepareRunInstances, which caps the launch at the available +// slots and returns ReservationCapacityExceeded when none remain. +func (rm *ResourceManager) ValidateReservationTarget(crID, accountID string, it *ec2.InstanceTypeInfo) error { + rm.mu.RLock() + defer rm.mu.RUnlock() + rec, ok := rm.reservations[crID] + if !ok || rec.AccountID != accountID { + return errors.New(awserrors.ErrorInvalidCapacityReservationIdNotFound) + } + if rec.InstanceType != aws.StringValue(it.InstanceType) { + return errors.New(awserrors.ErrorInvalidParameterValue) + } + return nil +} + +// ListReservations returns value-copy snapshots of the account's reservations on +// this node, taken under the read lock. Copies rather than the live pointers +// because ConsumedCount is mutated under rm.mu by the launch/release paths, so a +// caller reading the shared record lock-free would race. +func (rm *ResourceManager) ListReservations(accountID string) []capacityReservation { rm.mu.RLock() defer rm.mu.RUnlock() - var out []*capacityReservation + var out []capacityReservation for _, rec := range rm.reservations { if rec.AccountID == accountID { - out = append(out, rec) + out = append(out, *rec) } } return out diff --git a/spinifex/daemon/capacityreservation_test.go b/spinifex/daemon/capacityreservation_test.go index 451e3a6de..46d39f8fa 100644 --- a/spinifex/daemon/capacityreservation_test.go +++ b/spinifex/daemon/capacityreservation_test.go @@ -169,3 +169,187 @@ func TestCapacityReservation_ConcurrentNoOvercommit(t *testing.T) { assert.Equal(t, sumVCPU, rm.reservedCRVCPU) assert.Equal(t, sumMem, rm.reservedCRMem) } + +// AllocateFromReservation moves a slot reservedCR*->allocated* without touching +// schedulable capacity, bumps ConsumedCount, and drops AvailableInstanceCount. +func TestAllocateFromReservation_NetZeroSwap(t *testing.T) { + rm := newReservationTestRM() + mt := microType() + rec := microReservation("cr-1", "acct-a", 2) + require.NoError(t, rm.CreateReservation(rec)) + + before := rm.canAllocate(mt, 100) + require.Equal(t, 2, before, "carve-out of 2 leaves room for 2 general launches") + + require.NoError(t, rm.AllocateFromReservation("cr-1", "acct-a", mt)) + + rm.mu.RLock() + assert.Equal(t, 2, rm.reservedCRVCPU, "one slot left reservedCR*") + assert.Equal(t, 1.0, rm.reservedCRMem) + assert.Equal(t, 2, rm.allocatedVCPU, "consumed slot now in allocated*") + assert.Equal(t, 1.0, rm.allocatedMem) + assert.Equal(t, 1, rec.ConsumedCount) + rm.mu.RUnlock() + + assert.Equal(t, before, rm.canAllocate(mt, 100), "schedulable capacity unchanged by the swap") + assert.Equal(t, int64(1), aws.Int64Value(rec.toAWSCapacityReservation().AvailableInstanceCount)) +} + +// The semantic checks reject unknown/foreign ids, type mismatch, and over-consume +// with their mapped AWS error codes, leaving accounting untouched. +func TestAllocateFromReservation_Errors(t *testing.T) { + rm := newReservationTestRM() + mt := microType() + require.NoError(t, rm.CreateReservation(microReservation("cr-1", "acct-a", 1))) + + err := rm.AllocateFromReservation("cr-missing", "acct-a", mt) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidCapacityReservationIdNotFound, err.Error()) + + err = rm.AllocateFromReservation("cr-1", "acct-b", mt) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidCapacityReservationIdNotFound, err.Error(), "foreign account is not found") + + wrongType := &ec2.InstanceTypeInfo{ + InstanceType: aws.String("t3.large"), + VCpuInfo: &ec2.VCpuInfo{DefaultVCpus: aws.Int64(2)}, + MemoryInfo: &ec2.MemoryInfo{SizeInMiB: aws.Int64(8192)}, + } + err = rm.AllocateFromReservation("cr-1", "acct-a", wrongType) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error()) + + require.NoError(t, rm.AllocateFromReservation("cr-1", "acct-a", mt), "first slot consumes") + err = rm.AllocateFromReservation("cr-1", "acct-a", mt) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorReservationCapacityExceeded, err.Error(), "second exceeds Total=1") +} + +// ReleaseToReservation is the inverse swap when the reservation still exists: +// allocated*->reservedCR*, ConsumedCount--, AvailableInstanceCount restored. +func TestReleaseToReservation_RestoresSlot(t *testing.T) { + rm := newReservationTestRM() + mt := microType() + rec := microReservation("cr-1", "acct-a", 2) + require.NoError(t, rm.CreateReservation(rec)) + require.NoError(t, rm.AllocateFromReservation("cr-1", "acct-a", mt)) + + rm.ReleaseToReservation("cr-1", mt) + + rm.mu.RLock() + assert.Equal(t, 4, rm.reservedCRVCPU, "slot returned to reservedCR*") + assert.Equal(t, 2.0, rm.reservedCRMem) + assert.Equal(t, 0, rm.allocatedVCPU) + assert.Equal(t, 0.0, rm.allocatedMem) + assert.Equal(t, 0, rec.ConsumedCount) + rm.mu.RUnlock() + + assert.Equal(t, int64(2), aws.Int64Value(rec.toAWSCapacityReservation().AvailableInstanceCount)) +} + +// After the reservation is cancelled, a still-running consumer's release frees to +// the general pool (catalog charge) instead of a now-gone reservedCR*, never +// overcounting. Cancel itself frees only the unconsumed remainder. +func TestReleaseToReservation_AfterCancelFreesToGeneralPool(t *testing.T) { + rm := newReservationTestRM() + mt := microType() + require.NoError(t, rm.CreateReservation(microReservation("cr-1", "acct-a", 2))) + require.NoError(t, rm.AllocateFromReservation("cr-1", "acct-a", mt)) + + _, ok := rm.CancelReservation("cr-1", "acct-a") + require.True(t, ok) + + rm.mu.RLock() + assert.Equal(t, 0, rm.reservedCRVCPU, "cancel freed only the unconsumed slot") + assert.Equal(t, 0.0, rm.reservedCRMem) + assert.Equal(t, 2, rm.allocatedVCPU, "consumed slot keeps running on general capacity") + rm.mu.RUnlock() + + rm.ReleaseToReservation("cr-1", mt) + + rm.mu.RLock() + assert.Equal(t, 0, rm.allocatedVCPU, "release frees the dangling consumer to general pool") + assert.Equal(t, 0.0, rm.allocatedMem) + assert.Equal(t, 0, rm.reservedCRVCPU, "no phantom reservedCR* re-added") + rm.mu.RUnlock() + + assert.Equal(t, 4, rm.canAllocate(mt, 100), "general capacity fully restored") +} + +// ReservationAvailable reports Total-Consumed, and 0 for unknown/foreign-account +// /type-mismatch so the count site degrades to the daemon's up-front error. +func TestReservationAvailable(t *testing.T) { + rm := newReservationTestRM() + mt := microType() + require.NoError(t, rm.CreateReservation(microReservation("cr-1", "acct-a", 2))) + + assert.Equal(t, 2, rm.ReservationAvailable("cr-1", "acct-a", mt)) + assert.Equal(t, 0, rm.ReservationAvailable("cr-missing", "acct-a", mt)) + assert.Equal(t, 0, rm.ReservationAvailable("cr-1", "acct-b", mt), "foreign account sees nothing") + + wrongType := &ec2.InstanceTypeInfo{InstanceType: aws.String("t3.large")} + assert.Equal(t, 0, rm.ReservationAvailable("cr-1", "acct-a", wrongType), "type mismatch sees nothing") + + require.NoError(t, rm.AllocateFromReservation("cr-1", "acct-a", mt)) + assert.Equal(t, 1, rm.ReservationAvailable("cr-1", "acct-a", mt), "consume drops availability") +} + +// ValidateReservationTarget is the up-front semantic check: unknown/foreign ids +// map to NotFound, a type mismatch to InvalidParameterValue, and a matching +// reservation passes (the full case is left to the launch-side count gate). +func TestValidateReservationTarget(t *testing.T) { + rm := newReservationTestRM() + mt := microType() + require.NoError(t, rm.CreateReservation(microReservation("cr-1", "acct-a", 1))) + + require.NoError(t, rm.ValidateReservationTarget("cr-1", "acct-a", mt)) + + err := rm.ValidateReservationTarget("cr-missing", "acct-a", mt) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidCapacityReservationIdNotFound, err.Error()) + + err = rm.ValidateReservationTarget("cr-1", "acct-b", mt) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidCapacityReservationIdNotFound, err.Error(), "foreign account is not found") + + wrongType := &ec2.InstanceTypeInfo{InstanceType: aws.String("t3.large")} + err = rm.ValidateReservationTarget("cr-1", "acct-a", wrongType) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error()) + + // A full reservation still validates — the count gate handles capacity. + require.NoError(t, rm.AllocateFromReservation("cr-1", "acct-a", mt)) + require.NoError(t, rm.ValidateReservationTarget("cr-1", "acct-a", mt), "full reservation passes the semantic check") +} + +// Concurrent general allocate and AllocateFromReservation share rm.mu; the swap is +// net-zero on reservedCR*+allocated*, so committed compute never exceeds the host +// and ConsumedCount never passes Total. +func TestAllocateFromReservation_ConcurrentNoOvercommit(t *testing.T) { + rm := newReservationTestRM() + mt := microType() + rec := microReservation("cr-1", "acct-a", 2) + require.NoError(t, rm.CreateReservation(rec)) + + const workers = 32 + var wg sync.WaitGroup + wg.Add(workers) + for i := range workers { + go func(i int) { + defer wg.Done() + if i%2 == 0 { + _ = rm.AllocateFromReservation("cr-1", "acct-a", mt) + } else { + _ = rm.allocate(mt) + } + }(i) + } + wg.Wait() + + rm.mu.RLock() + defer rm.mu.RUnlock() + assert.LessOrEqual(t, rm.reservedCRVCPU+rm.allocatedVCPU, rm.hostVCPU, "vCPU never overcommitted") + assert.LessOrEqual(t, rm.reservedCRMem+rm.allocatedMem, rm.hostMemGB, "memory never overcommitted") + assert.LessOrEqual(t, rec.ConsumedCount, rec.TotalInstanceCount, "ConsumedCount never passes Total") + assert.GreaterOrEqual(t, rec.ConsumedCount, 0) +} diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 879848d1f..ad0f58e7d 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -2141,6 +2141,8 @@ func (rm *ResourceManager) deallocate(instanceType *ec2.InstanceTypeInfo) { rm.updateInstanceSubscriptions() } +var _ handlers_ec2_instance.InstanceTypeAllocator = (*ResourceManager)(nil) + // Allocate, Deallocate, CanAllocate satisfy handlers_ec2_instance.InstanceTypeAllocator. func (rm *ResourceManager) Allocate(it *ec2.InstanceTypeInfo) error { return rm.allocate(it) } func (rm *ResourceManager) Deallocate(it *ec2.InstanceTypeInfo) { rm.deallocate(it) } diff --git a/spinifex/daemon/daemon_handlers_capacityreservation.go b/spinifex/daemon/daemon_handlers_capacityreservation.go index 8ec587008..1543ec374 100644 --- a/spinifex/daemon/daemon_handlers_capacityreservation.go +++ b/spinifex/daemon/daemon_handlers_capacityreservation.go @@ -50,7 +50,7 @@ func (d *Daemon) handleEC2CreateCapacityReservation(msg *nats.Msg) { AvailabilityZone: aws.StringValue(input.AvailabilityZone), TotalInstanceCount: int(aws.Int64Value(input.InstanceCount)), VCPUPerInstance: int(instanceTypeVCPUs(it)), - MemGBPerInstance: float64(instanceTypeMemoryMiB(it)) / 1024.0, + MemGBPerInstance: float64(d.resourceMgr.instanceMemChargeMiB(it)) / 1024.0, InstanceMatchCriteria: matchCriteria, Tenancy: tenancy, InstancePlatform: aws.StringValue(input.InstancePlatform), @@ -62,6 +62,19 @@ func (d *Daemon) handleEC2CreateCapacityReservation(msg *nats.Msg) { return } + // Subscribe the per-reservation launch subject before responding: the SUB and + // this reply share d.natsConn, so the server registers the responder before + // the gateway sees the new id and can route a targeted launch to it. A failed + // subscribe would leave the reservation holding capacity no targeted launch + // could ever reach, so roll it back rather than return a dead id. + if err := d.subscribeReservationLaunch(rec.ID); err != nil { + slog.Error("Failed to subscribe reservation launch subject, rolling back reservation", + "crId", rec.ID, "err", err) + d.resourceMgr.CancelReservation(rec.ID, accountID) + respondWithError(msg, awserrors.ErrorServerInternal) + return + } + respondWithJSON(msg, rec.toAWSCapacityReservation()) } @@ -92,22 +105,71 @@ func (d *Daemon) handleEC2CancelCapacityReservation(msg *nats.Msg) { } _, found := d.resourceMgr.CancelReservation(aws.StringValue(input.CapacityReservationId), accountID) + if found { + // Drop the launch subject so a targeted launch against the cancelled id + // hits no responder → InvalidCapacityReservationId.NotFound at the gateway. + d.unsubscribeReservationLaunch(aws.StringValue(input.CapacityReservationId)) + } respondWithJSON(msg, &ec2.CancelCapacityReservationOutput{Return: aws.Bool(found)}) } +// reservationLaunchSubject is the per-reservation subject the owning daemon +// subscribes so the gateway routes a targeted launch straight to it, with no +// cr→node lookup. ErrNoResponders on this subject therefore means the +// reservation is gone (cancelled or lost to a restart). +func reservationLaunchSubject(crID string) string { + return "ec2.RunInstances.cr." + crID +} + +// subscribeReservationLaunch wires the cr launch subject to the standard +// RunInstances handler — the target id rides in the input, so handleEC2RunInstances +// serves it unchanged. Tracked in natsSubscriptions (keyed by cr id) so cancel +// and shutdown can drop it. Idempotent; returns the subscribe error so the +// caller can roll the reservation back rather than leave it unlaunchable. +func (d *Daemon) subscribeReservationLaunch(crID string) error { + if d.natsConn == nil { + return nil + } + d.mu.Lock() + defer d.mu.Unlock() + if _, exists := d.natsSubscriptions[crID]; exists { + return nil + } + sub, err := d.natsConn.Subscribe(reservationLaunchSubject(crID), d.handleEC2RunInstances) + if err != nil { + return err + } + d.natsSubscriptions[crID] = sub + return nil +} + +// unsubscribeReservationLaunch drops the cr launch subscription on cancel. +func (d *Daemon) unsubscribeReservationLaunch(crID string) { + d.mu.Lock() + sub, ok := d.natsSubscriptions[crID] + if ok { + delete(d.natsSubscriptions, crID) + } + d.mu.Unlock() + if ok { + if err := sub.Unsubscribe(); err != nil { + slog.Warn("Failed to unsubscribe reservation launch subject", "crId", crID, "err", err) + } + } +} + // toAWSCapacityReservation renders the in-memory record as the AWS API type. -// Reservations are always active with AvailableInstanceCount equal to the total, +// Reservations are always active with AvailableInstanceCount = Total − Consumed, // and never expire (EndDateType=unlimited). func (r *capacityReservation) toAWSCapacityReservation() *ec2.CapacityReservation { - count := int64(r.TotalInstanceCount) return &ec2.CapacityReservation{ CapacityReservationId: aws.String(r.ID), OwnerId: aws.String(r.AccountID), InstanceType: aws.String(r.InstanceType), InstancePlatform: aws.String(r.InstancePlatform), AvailabilityZone: aws.String(r.AvailabilityZone), - TotalInstanceCount: aws.Int64(count), - AvailableInstanceCount: aws.Int64(count), + TotalInstanceCount: aws.Int64(int64(r.TotalInstanceCount)), + AvailableInstanceCount: aws.Int64(int64(r.TotalInstanceCount - r.ConsumedCount)), State: aws.String(ec2.CapacityReservationStateActive), StartDate: aws.Time(r.CreateDate), CreateDate: aws.Time(r.CreateDate), diff --git a/spinifex/daemon/daemon_handlers_capacityreservation_test.go b/spinifex/daemon/daemon_handlers_capacityreservation_test.go index 91495a78d..1a90cfcb2 100644 --- a/spinifex/daemon/daemon_handlers_capacityreservation_test.go +++ b/spinifex/daemon/daemon_handlers_capacityreservation_test.go @@ -249,3 +249,61 @@ func TestHandleEC2CancelCapacityReservation_ScopingAndAck(t *testing.T) { assert.False(t, cancel(testAccountID), "a second cancel reports not-found") } + +// Create subscribes the per-reservation launch subject (so a targeted launch has +// a responder) and Cancel drops it (so a launch against the cancelled id returns +// ErrNoResponders → InvalidCapacityReservationId.NotFound at the gateway). +func TestHandleEC2CapacityReservation_LaunchSubjectLifecycle(t *testing.T) { + daemon := createTestDaemon(t, sharedNATSURL) + + instType := getTestInstanceType(t) + it := daemon.resourceMgr.instanceTypes[instType] + require.NotNil(t, it) + require.GreaterOrEqual(t, daemon.resourceMgr.canAllocate(it, 1000), 1) + + createSubject := "ec2.CreateCapacityReservation." + daemon.node + createSub, err := daemon.natsConn.Subscribe(createSubject, daemon.handleEC2CreateCapacityReservation) + require.NoError(t, err) + defer func() { _ = createSub.Unsubscribe() }() + cancelSub, err := daemon.natsConn.Subscribe("ec2.CancelCapacityReservation", daemon.handleEC2CancelCapacityReservation) + require.NoError(t, err) + defer func() { _ = cancelSub.Unsubscribe() }() + + reqData, _ := json.Marshal(&ec2.CreateCapacityReservationInput{ + InstanceType: aws.String(instType), InstanceCount: aws.Int64(1), + AvailabilityZone: aws.String("ap-southeast-2a"), InstancePlatform: aws.String("Linux/UNIX"), + }) + reply, err := natsRequest(daemon.natsConn, createSubject, reqData, 5*time.Second) + require.NoError(t, err) + require.Empty(t, replyErrCode(t, reply.Data), "create should not error: %s", reply.Data) + var cr ec2.CapacityReservation + require.NoError(t, json.Unmarshal(reply.Data, &cr)) + crID := aws.StringValue(cr.CapacityReservationId) + + daemon.mu.Lock() + _, tracked := daemon.natsSubscriptions[crID] + daemon.mu.Unlock() + assert.True(t, tracked, "create tracks the cr launch subscription") + + // An empty payload makes handleEC2RunInstances reject on MissingParameter + // before any allocation; the point is only that a responder exists (no + // ErrNoResponders) — the SUB raced ahead of the create reply on the same conn. + launchSubject := reservationLaunchSubject(crID) + _, err = natsRequest(daemon.natsConn, launchSubject, []byte("{}"), 2*time.Second) + require.NotErrorIs(t, err, nats.ErrNoResponders, "cr launch subject has a responder after create") + + cancelData, _ := json.Marshal(&ec2.CancelCapacityReservationInput{CapacityReservationId: aws.String(crID)}) + creply, err := natsRequest(daemon.natsConn, "ec2.CancelCapacityReservation", cancelData, 5*time.Second) + require.NoError(t, err) + var cout ec2.CancelCapacityReservationOutput + require.NoError(t, json.Unmarshal(creply.Data, &cout)) + require.True(t, aws.BoolValue(cout.Return), "owner cancel succeeds") + + daemon.mu.Lock() + _, tracked = daemon.natsSubscriptions[crID] + daemon.mu.Unlock() + assert.False(t, tracked, "cancel drops the cr launch subscription") + + _, err = natsRequest(daemon.natsConn, launchSubject, []byte("{}"), 1*time.Second) + assert.ErrorIs(t, err, nats.ErrNoResponders, "cr launch subject has no responder after cancel") +} diff --git a/spinifex/daemon/daemon_handlers_instance.go b/spinifex/daemon/daemon_handlers_instance.go index de87ee6b3..fdf7ac44e 100644 --- a/spinifex/daemon/daemon_handlers_instance.go +++ b/spinifex/daemon/daemon_handlers_instance.go @@ -50,7 +50,21 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { return } - reservation, instances, instanceType, err := d.instanceService.PrepareRunInstances(input, accountID) + // Targeted launch: the gateway routes only when an explicit reservation id is + // present, but the id rides in the input either way. Validate semantics + // up-front (the per-instance atomic re-check under rm.mu covers the race); + // the count gate in PrepareRunInstances handles a full reservation. + reservationID := capacityReservationTargetID(input) + if reservationID != "" { + if it, ok := d.resourceMgr.instanceTypes[aws.StringValue(input.InstanceType)]; ok { + if vErr := d.resourceMgr.ValidateReservationTarget(reservationID, accountID, it); vErr != nil { + respondWithError(msg, awserrors.ValidErrorCode(vErr.Error())) + return + } + } + } + + reservation, instances, instanceType, err := d.instanceService.PrepareRunInstances(input, accountID, reservationID) if err != nil { respondWithError(msg, awserrors.ValidErrorCode(err.Error())) return @@ -68,7 +82,11 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { slog.Error("handleEC2RunInstances failed to marshal reservation", "err", err) respondWithError(msg, awserrors.ErrorServerInternal) for range instances { - d.resourceMgr.deallocate(instanceType) + if reservationID == "" { + d.resourceMgr.deallocate(instanceType) + } else { + d.resourceMgr.ReleaseToReservation(reservationID, instanceType) + } } return } @@ -100,6 +118,16 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { d.instanceService.LaunchRunInstances(instances, input, instanceType) } +// capacityReservationTargetID returns the explicit targeted-launch reservation id +// from the input, or "" when the launch is untargeted (general path). +func capacityReservationTargetID(input *ec2.RunInstancesInput) string { + spec := input.CapacityReservationSpecification + if spec == nil || spec.CapacityReservationTarget == nil { + return "" + } + return aws.StringValue(spec.CapacityReservationTarget.CapacityReservationId) +} + // describeInstancesValidFilters defines the set of filter names accepted by DescribeInstances. var describeInstancesValidFilters = map[string]bool{ "instance-state-name": true, @@ -307,6 +335,19 @@ func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) { } } + // Echo the consumed capacity reservation so targeted-launch + // Terraform converges — without it the instance reports no + // reservation and the plan never settles. + if instance.CapacityReservationId != "" { + instanceCopy.CapacityReservationId = aws.String(instance.CapacityReservationId) + instanceCopy.CapacityReservationSpecification = &ec2.CapacityReservationSpecificationResponse{ + CapacityReservationPreference: aws.String(ec2.CapacityReservationPreferenceOpen), + CapacityReservationTarget: &ec2.CapacityReservationTargetResponse{ + CapacityReservationId: aws.String(instance.CapacityReservationId), + }, + } + } + // Apply filters against the fully-built instance copy if len(parsedFilters) > 0 && !instanceMatchesFilters(instance, &instanceCopy, parsedFilters) { continue diff --git a/spinifex/daemon/daemon_handlers_instance_test.go b/spinifex/daemon/daemon_handlers_instance_test.go index 857dddbfc..2196c778b 100644 --- a/spinifex/daemon/daemon_handlers_instance_test.go +++ b/spinifex/daemon/daemon_handlers_instance_test.go @@ -595,3 +595,82 @@ func TestHandleEC2DescribeStoppedInstances_ReservationGrouping(t *testing.T) { assert.Equal(t, "r-shared", *output.Reservations[0].ReservationId) assert.Len(t, output.Reservations[0].Instances, 2) } + +// DescribeInstances echoes the consumed capacity reservation so a targeted +// launch reports a CapacityReservationId (without it, Terraform never converges). +func TestHandleEC2DescribeInstances_CapacityReservationEcho(t *testing.T) { + daemon := createTestDaemon(t, sharedNATSURL) + + reservation := &ec2.Reservation{} + reservation.SetReservationId("r-cr-echo") + reservation.SetOwnerId(testAccountID) + instance := &ec2.Instance{} + instance.SetInstanceId("i-cr-echo") + instance.SetInstanceType("t3.micro") + + daemon.vmMgr.Insert(&vm.VM{ + ID: "i-cr-echo", + Status: vm.StateRunning, + AccountID: testAccountID, + CapacityReservationId: "cr-0123456789abcdef0", + Reservation: reservation, + Instance: instance, + }) + + sub, err := daemon.natsConn.Subscribe("ec2.DescribeInstances", daemon.handleEC2DescribeInstances) + require.NoError(t, err) + defer func() { _ = sub.Unsubscribe() }() + + reqData, _ := json.Marshal(&ec2.DescribeInstancesInput{}) + reply, err := natsRequest(daemon.natsConn, "ec2.DescribeInstances", reqData, 5*time.Second) + require.NoError(t, err) + var out ec2.DescribeInstancesOutput + require.NoError(t, json.Unmarshal(reply.Data, &out)) + require.Len(t, out.Reservations, 1) + require.Len(t, out.Reservations[0].Instances, 1) + + got := out.Reservations[0].Instances[0] + assert.Equal(t, "cr-0123456789abcdef0", aws.StringValue(got.CapacityReservationId)) + require.NotNil(t, got.CapacityReservationSpecification) + assert.Equal(t, ec2.CapacityReservationPreferenceOpen, + aws.StringValue(got.CapacityReservationSpecification.CapacityReservationPreference)) + require.NotNil(t, got.CapacityReservationSpecification.CapacityReservationTarget) + assert.Equal(t, "cr-0123456789abcdef0", + aws.StringValue(got.CapacityReservationSpecification.CapacityReservationTarget.CapacityReservationId)) +} + +// An instance with no reservation reports no capacity-reservation fields. +func TestHandleEC2DescribeInstances_NoCapacityReservationEcho(t *testing.T) { + daemon := createTestDaemon(t, sharedNATSURL) + + reservation := &ec2.Reservation{} + reservation.SetReservationId("r-plain") + reservation.SetOwnerId(testAccountID) + instance := &ec2.Instance{} + instance.SetInstanceId("i-plain") + instance.SetInstanceType("t3.micro") + + daemon.vmMgr.Insert(&vm.VM{ + ID: "i-plain", + Status: vm.StateRunning, + AccountID: testAccountID, + Reservation: reservation, + Instance: instance, + }) + + sub, err := daemon.natsConn.Subscribe("ec2.DescribeInstances", daemon.handleEC2DescribeInstances) + require.NoError(t, err) + defer func() { _ = sub.Unsubscribe() }() + + reqData, _ := json.Marshal(&ec2.DescribeInstancesInput{}) + reply, err := natsRequest(daemon.natsConn, "ec2.DescribeInstances", reqData, 5*time.Second) + require.NoError(t, err) + var out ec2.DescribeInstancesOutput + require.NoError(t, json.Unmarshal(reply.Data, &out)) + require.Len(t, out.Reservations, 1) + require.Len(t, out.Reservations[0].Instances, 1) + + got := out.Reservations[0].Instances[0] + assert.Empty(t, aws.StringValue(got.CapacityReservationId)) + assert.Nil(t, got.CapacityReservationSpecification) +} diff --git a/spinifex/daemon/daemon_system_instance.go b/spinifex/daemon/daemon_system_instance.go index 5a38879ed..0447875e4 100644 --- a/spinifex/daemon/daemon_system_instance.go +++ b/spinifex/daemon/daemon_system_instance.go @@ -427,7 +427,7 @@ func (d *Daemon) launchAMISystemInstance(input *sysinstance.SystemInstanceInput) }} } - _, instances, instanceType, err := d.instanceService.PrepareRunInstances(runInput, input.AccountID) + _, instances, instanceType, err := d.instanceService.PrepareRunInstances(runInput, input.AccountID, "") if err != nil { return nil, err } diff --git a/spinifex/daemon/vm_adapters.go b/spinifex/daemon/vm_adapters.go index 6deb41f6f..3b9c90540 100644 --- a/spinifex/daemon/vm_adapters.go +++ b/spinifex/daemon/vm_adapters.go @@ -313,6 +313,14 @@ func (a *resourceControllerAdapter) Deallocate(instanceType string) { a.rm.deallocate(it) } +func (a *resourceControllerAdapter) ReleaseToReservation(reservationID, instanceType string) { + it := a.rm.instanceTypes[instanceType] + if it == nil { + return + } + a.rm.ReleaseToReservation(reservationID, it) +} + func (a *resourceControllerAdapter) CanAllocate(instanceType string, count int) int { it := a.rm.instanceTypes[instanceType] if it == nil { diff --git a/spinifex/gateway/ec2/instance/RunInstances.go b/spinifex/gateway/ec2/instance/RunInstances.go index 55b0cfda9..64fcf4b5a 100644 --- a/spinifex/gateway/ec2/instance/RunInstances.go +++ b/spinifex/gateway/ec2/instance/RunInstances.go @@ -104,6 +104,25 @@ func runInstancesInner(input *ec2.RunInstancesInput, natsConn *nats.Conn, iamSvc return reservation, err } + // Targeted capacity-reservation launch routes straight to the owning node's + // cr-subject. The gateway does input-only checks (malformed id, the PG+CR + // combination); the owning daemon does the semantic checks (account, type, + // full) only it can see, and ErrNoResponders surfaces a gone id as NotFound. + if crID := capacityReservationTargetID(input); crID != "" { + if !strings.HasPrefix(crID, "cr-") { + return reservation, errors.New(awserrors.ErrorInvalidCapacityReservationIdMalformed) + } + if placementGroupName(input) != "" { + return reservation, errors.New(awserrors.ErrorInvalidParameterValue) + } + reservationPtr, err := runIntoReservation(input, natsConn, accountID, crID) + if err != nil { + return reservation, err + } + enrichReservationWithProfileID(reservationPtr, resolvedProfile) + return *reservationPtr, nil + } + groupName := placementGroupName(input) if groupName != "" { strategy, err := lookupPlacementGroupStrategy(natsConn, accountID, groupName) @@ -198,6 +217,17 @@ func placementGroupName(input *ec2.RunInstancesInput) string { return "" } +// capacityReservationTargetID returns the explicit targeted-launch reservation id +// from the input, or "" when the launch is untargeted (general path). Preference +// is ignored — only a present target id routes to the cr-subject. +func capacityReservationTargetID(input *ec2.RunInstancesInput) string { + spec := input.CapacityReservationSpecification + if spec == nil || spec.CapacityReservationTarget == nil { + return "" + } + return aws.StringValue(spec.CapacityReservationTarget.CapacityReservationId) +} + // lookupPlacementGroupStrategy returns the strategy of a placement group, or an error if absent/unavailable. func lookupPlacementGroupStrategy(natsConn *nats.Conn, accountID, groupName string) (string, error) { pgSvc := handlers_ec2_placementgroup.NewNATSPlacementGroupService(natsConn) diff --git a/spinifex/gateway/ec2/instance/RunInstancesIntoReservation.go b/spinifex/gateway/ec2/instance/RunInstancesIntoReservation.go new file mode 100644 index 000000000..5a767ac14 --- /dev/null +++ b/spinifex/gateway/ec2/instance/RunInstancesIntoReservation.go @@ -0,0 +1,28 @@ +package gateway_ec2_instance + +import ( + "errors" + "time" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" +) + +// runIntoReservation routes a targeted launch straight to the owning daemon's +// per-reservation subject. Only the owner subscribes ec2.RunInstances.cr., +// so NATS addresses it without any cr→node lookup. ErrNoResponders means the +// reservation is gone (cancelled or lost to a restart) → NotFound; any semantic +// error (type mismatch, full) rides back as the daemon's awserror code. +func runIntoReservation(input *ec2.RunInstancesInput, natsConn *nats.Conn, accountID, crID string) (*ec2.Reservation, error) { + subject := "ec2.RunInstances.cr." + crID + reservation, err := utils.NATSRequest[ec2.Reservation](natsConn, subject, input, 5*time.Minute, accountID) + if err != nil { + if errors.Is(err, nats.ErrNoResponders) { + return nil, errors.New(awserrors.ErrorInvalidCapacityReservationIdNotFound) + } + return nil, err + } + return reservation, nil +} diff --git a/spinifex/gateway/ec2/instance/RunInstancesIntoReservation_test.go b/spinifex/gateway/ec2/instance/RunInstancesIntoReservation_test.go new file mode 100644 index 000000000..3ec6e6992 --- /dev/null +++ b/spinifex/gateway/ec2/instance/RunInstancesIntoReservation_test.go @@ -0,0 +1,124 @@ +package gateway_ec2_instance + +import ( + "encoding/json" + "testing" + "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/utils" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const testCRID = "cr-0123456789abcdef0" + +// crTargetedInput clones the valid defaults and targets a capacity reservation. +func crTargetedInput(crID string) *ec2.RunInstancesInput { + in := defaults + in.CapacityReservationSpecification = &ec2.CapacityReservationSpecification{ + CapacityReservationTarget: &ec2.CapacityReservationTarget{ + CapacityReservationId: aws.String(crID), + }, + } + return &in +} + +func TestCapacityReservationTargetID(t *testing.T) { + assert.Empty(t, capacityReservationTargetID(&ec2.RunInstancesInput{}), "no spec → general path") + assert.Empty(t, capacityReservationTargetID(&ec2.RunInstancesInput{ + CapacityReservationSpecification: &ec2.CapacityReservationSpecification{ + CapacityReservationPreference: aws.String(ec2.CapacityReservationPreferenceOpen), + }, + }), "open without target → general path") + assert.Equal(t, testCRID, capacityReservationTargetID(crTargetedInput(testCRID))) +} + +// A malformed reservation id is rejected at the gateway before any NATS call. +func TestRunInstancesInner_TargetedMalformedID(t *testing.T) { + _, err := runInstancesInner(crTargetedInput("bogus-id"), nil, nil, "123456789012", nil, 1) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidCapacityReservationIdMalformed, err.Error()) +} + +// The placement-group + capacity-reservation combination is rejected (Phase 2). +func TestRunInstancesInner_TargetedWithPlacementGroup(t *testing.T) { + input := crTargetedInput(testCRID) + input.Placement = &ec2.Placement{GroupName: aws.String("pg-cluster")} + _, err := runInstancesInner(input, nil, nil, "123456789012", nil, 1) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error()) +} + +// No responder on the cr-subject (cancelled / restarted reservation) maps the +// transport-level ErrNoResponders to InvalidCapacityReservationId.NotFound. +func TestRunIntoReservation_NoResponder(t *testing.T) { + _, nc := startTestNATSServer(t) + _, err := runIntoReservation(crTargetedInput(testCRID), nc, "123456789012", testCRID) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorInvalidCapacityReservationIdNotFound, err.Error()) +} + +// A targeted launch routes to ec2.RunInstances.cr. and returns the owning +// daemon's reservation verbatim. +func TestRunIntoReservation_Success(t *testing.T) { + _, nc := startTestNATSServer(t) + + sub, err := nc.Subscribe("ec2.RunInstances.cr."+testCRID, func(msg *nats.Msg) { + data, _ := json.Marshal(ec2.Reservation{ + ReservationId: aws.String("r-cr"), + Instances: []*ec2.Instance{{InstanceId: aws.String("i-cr1")}}, + }) + _ = msg.Respond(data) + }) + require.NoError(t, err) + defer func() { _ = sub.Unsubscribe() }() + time.Sleep(50 * time.Millisecond) + + reservation, err := runIntoReservation(crTargetedInput(testCRID), nc, "123456789012", testCRID) + require.NoError(t, err) + require.Len(t, reservation.Instances, 1) + assert.Equal(t, "i-cr1", aws.StringValue(reservation.Instances[0].InstanceId)) +} + +// A semantic error from the owning daemon (e.g. a full reservation) rides back +// as its awserror code, not the generic NotFound. +func TestRunIntoReservation_DaemonErrorPropagates(t *testing.T) { + _, nc := startTestNATSServer(t) + + sub, err := nc.Subscribe("ec2.RunInstances.cr."+testCRID, func(msg *nats.Msg) { + _ = msg.Respond(utils.GenerateErrorPayload(awserrors.ErrorReservationCapacityExceeded)) + }) + require.NoError(t, err) + defer func() { _ = sub.Unsubscribe() }() + time.Sleep(50 * time.Millisecond) + + _, err = runIntoReservation(crTargetedInput(testCRID), nc, "123456789012", testCRID) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorReservationCapacityExceeded, err.Error()) +} + +// The full runInstancesInner branch routes a targeted launch to the cr-subject +// (not the general distribute path) and enriches the result. +func TestRunInstancesInner_TargetedRoutesToCRSubject(t *testing.T) { + _, nc := startTestNATSServer(t) + + sub, err := nc.Subscribe("ec2.RunInstances.cr."+testCRID, func(msg *nats.Msg) { + data, _ := json.Marshal(ec2.Reservation{ + ReservationId: aws.String("r-cr"), + Instances: []*ec2.Instance{{InstanceId: aws.String("i-cr1")}}, + }) + _ = msg.Respond(data) + }) + require.NoError(t, err) + defer func() { _ = sub.Unsubscribe() }() + time.Sleep(50 * time.Millisecond) + + reservation, err := runInstancesInner(crTargetedInput(testCRID), nc, nil, "123456789012", nil, 1) + require.NoError(t, err) + require.Len(t, reservation.Instances, 1) + assert.Equal(t, "i-cr1", aws.StringValue(reservation.Instances[0].InstanceId)) +} diff --git a/spinifex/handlers/ec2/instance/service.go b/spinifex/handlers/ec2/instance/service.go index 2b6e98700..35f49fbb7 100644 --- a/spinifex/handlers/ec2/instance/service.go +++ b/spinifex/handlers/ec2/instance/service.go @@ -59,6 +59,9 @@ type InstanceTypeAllocator interface { Allocate(instanceType *ec2.InstanceTypeInfo) error Deallocate(instanceType *ec2.InstanceTypeInfo) CanAllocate(instanceType *ec2.InstanceTypeInfo, count int) int + AllocateFromReservation(reservationID, accountID string, instanceType *ec2.InstanceTypeInfo) error + ReleaseToReservation(reservationID string, instanceType *ec2.InstanceTypeInfo) + ReservationAvailable(reservationID, accountID string, instanceType *ec2.InstanceTypeInfo) int InstanceTypes() map[string]*ec2.InstanceTypeInfo } diff --git a/spinifex/handlers/ec2/instance/service_impl.go b/spinifex/handlers/ec2/instance/service_impl.go index ae55c7bfc..be5488198 100644 --- a/spinifex/handlers/ec2/instance/service_impl.go +++ b/spinifex/handlers/ec2/instance/service_impl.go @@ -572,7 +572,13 @@ func instanceTagsFromSpec(specs []*ec2.TagSpecification) []*ec2.Tag { // PrepareRunInstances validates input, allocates capacity, creates VM metadata, // auto-creates the primary ENI, and auto-assigns a public IP when needed. // Does NOT touch vmMgr or NATS — callers insert VMs then call LaunchRunInstances. -func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, []*vm.VM, *ec2.InstanceTypeInfo, error) { +// +// A non-empty reservationID makes this a targeted launch into an On-Demand +// Capacity Reservation: capacity is consumed from that reservation (net-zero +// swap), the count is capped at the reservation's free slots (never spilling +// onto general capacity), and every rollback site returns slots to the +// reservation rather than the general pool. +func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput, accountID, reservationID string) (*ec2.Reservation, []*vm.VM, *ec2.InstanceTypeInfo, error) { if accountID == "" { return nil, nil, nil, errors.New(awserrors.ErrorServerInternal) } @@ -620,10 +626,21 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput, minCount := int(*input.MinCount) maxCount := int(*input.MaxCount) - allocatableCount := s.resourceMgr.CanAllocate(instanceType, maxCount) + var allocatableCount int + if reservationID == "" { + allocatableCount = s.resourceMgr.CanAllocate(instanceType, maxCount) + } else { + // Targeted launch: confined to the reservation. Cap at its free slots so + // the overflow never spills onto the node's general capacity. + allocatableCount = min(s.resourceMgr.ReservationAvailable(reservationID, accountID, 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) + errCode := awserrors.ErrorInsufficientInstanceCapacity + if reservationID != "" { + errCode = awserrors.ErrorReservationCapacityExceeded + } + slog.Error("PrepareRunInstances: insufficient capacity", "requested", minCount, "available", allocatableCount, "InstanceType", *input.InstanceType, "reservationId", reservationID) + return nil, nil, nil, errors.New(errCode) } launchCount := allocatableCount @@ -633,18 +650,32 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput, // allocatableCount — the < minCount branch below rolls back. 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) + var allocErr error + if reservationID == "" { + allocErr = s.resourceMgr.Allocate(instanceType) + } else { + allocErr = s.resourceMgr.AllocateFromReservation(reservationID, accountID, instanceType) + } + if allocErr != nil { + slog.Error("PrepareRunInstances: allocate failed mid-allocation", "allocated", allocatedCount, "err", allocErr) break } allocatedCount++ } if allocatedCount < minCount { for i := 0; i < allocatedCount; i++ { - s.resourceMgr.Deallocate(instanceType) + if reservationID == "" { + s.resourceMgr.Deallocate(instanceType) + } else { + s.resourceMgr.ReleaseToReservation(reservationID, instanceType) + } + } + errCode := awserrors.ErrorInsufficientInstanceCapacity + if reservationID != "" { + errCode = awserrors.ErrorReservationCapacityExceeded } slog.Error("PrepareRunInstances: insufficient capacity after allocation", "allocated", allocatedCount, "minCount", minCount) - return nil, nil, nil, errors.New(awserrors.ErrorInsufficientInstanceCapacity) + return nil, nil, nil, errors.New(errCode) } launchCount = allocatedCount @@ -657,7 +688,11 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput, if err != nil { slog.Error("PrepareRunInstances: RunInstance failed", "index", i, "err", err) lastRunErr = err - s.resourceMgr.Deallocate(instanceType) + if reservationID == "" { + s.resourceMgr.Deallocate(instanceType) + } else { + s.resourceMgr.ReleaseToReservation(reservationID, instanceType) + } continue } instance.BootMode = amiMeta.BootMode @@ -685,14 +720,22 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput, slog.Error("PrepareRunInstances: pre-created ENI specified but ENI service unavailable", "instanceId", instance.ID, "eniId", preExistingENIID) lastRunErr = errors.New(awserrors.ErrorServerInternal) - s.resourceMgr.Deallocate(instanceType) + if reservationID == "" { + s.resourceMgr.Deallocate(instanceType) + } else { + s.resourceMgr.ReleaseToReservation(reservationID, instanceType) + } continue } if err := s.attachPreCreatedENI(accountID, preExistingENIID, instance, ec2Instance); err != nil { slog.Error("PrepareRunInstances: pre-created ENI attach failed", "instanceId", instance.ID, "eniId", preExistingENIID, "err", err) lastRunErr = err - s.resourceMgr.Deallocate(instanceType) + if reservationID == "" { + s.resourceMgr.Deallocate(instanceType) + } else { + s.resourceMgr.ReleaseToReservation(reservationID, instanceType) + } continue } ec2Instance.SetAmiLaunchIndex(int64(len(allEC2Instances))) @@ -718,7 +761,11 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput, if eniErr != nil { slog.Error("PrepareRunInstances: auto-create ENI failed", "instanceId", instance.ID, "subnetId", *input.SubnetId, "err", eniErr) lastRunErr = eniErr - s.resourceMgr.Deallocate(instanceType) + if reservationID == "" { + s.resourceMgr.Deallocate(instanceType) + } else { + s.resourceMgr.ReleaseToReservation(reservationID, instanceType) + } continue } @@ -738,7 +785,11 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput, "eniId", instance.ENIId, "err", delErr) } lastRunErr = attachErr - s.resourceMgr.Deallocate(instanceType) + if reservationID == "" { + s.resourceMgr.Deallocate(instanceType) + } else { + s.resourceMgr.ReleaseToReservation(reservationID, instanceType) + } continue } ec2Instance.SetPrivateIpAddress(*eni.PrivateIpAddress) @@ -794,7 +845,11 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput, "eniId", *eni.NetworkInterfaceId, "err", delErr) } lastRunErr = errors.New(awserrors.ErrorInsufficientAddressCapacity) - s.resourceMgr.Deallocate(instanceType) + if reservationID == "" { + s.resourceMgr.Deallocate(instanceType) + } else { + s.resourceMgr.ReleaseToReservation(reservationID, instanceType) + } continue } else { if updateErr := s.eniCreator.UpdateENIPublicIP(accountID, *eni.NetworkInterfaceId, publicIP, poolName); updateErr != nil { @@ -808,7 +863,11 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput, utils.PublishNATEvent(s.natsConn, "vpc.delete-nat", *eni.VpcId, publicIP, *eni.PrivateIpAddress, portName, *eni.MacAddress) s.rollbackAutoAssignedPublicIP(accountID, instance.ID, *eni.NetworkInterfaceId, publicIP, poolName) lastRunErr = natErr - s.resourceMgr.Deallocate(instanceType) + if reservationID == "" { + s.resourceMgr.Deallocate(instanceType) + } else { + s.resourceMgr.ReleaseToReservation(reservationID, instanceType) + } continue } ec2Instance.PublicIpAddress = aws.String(publicIP) @@ -832,7 +891,11 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput, if len(instances) < minCount { for range instances { - s.resourceMgr.Deallocate(instanceType) + if reservationID == "" { + s.resourceMgr.Deallocate(instanceType) + } else { + s.resourceMgr.ReleaseToReservation(reservationID, instanceType) + } } errCode := awserrors.ErrorServerInternal if lastRunErr != nil { @@ -852,6 +915,9 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput, for _, instance := range instances { instance.Reservation = reservation instance.AccountID = accountID + if reservationID != "" { + instance.CapacityReservationId = reservationID + } if input.Placement != nil && input.Placement.GroupName != nil && *input.Placement.GroupName != "" { instance.PlacementGroupName = *input.Placement.GroupName } @@ -987,7 +1053,7 @@ func (s *InstanceServiceImpl) LaunchRunInstances(instances []*vm.VM, input *ec2. // RunInstances is for non-daemon callers (tests). The daemon calls // PrepareRunInstances + LaunchRunInstances directly to respond before launching. func (s *InstanceServiceImpl) RunInstances(input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, error) { - reservation, instances, instanceType, err := s.PrepareRunInstances(input, accountID) + reservation, instances, instanceType, err := s.PrepareRunInstances(input, accountID, "") if err != nil { return nil, err } diff --git a/spinifex/handlers/ec2/instance/service_impl_test.go b/spinifex/handlers/ec2/instance/service_impl_test.go index 04343a1e9..133d6f8f8 100644 --- a/spinifex/handlers/ec2/instance/service_impl_test.go +++ b/spinifex/handlers/ec2/instance/service_impl_test.go @@ -972,6 +972,11 @@ type fakeResourceCapacityProvider struct { allocated []*ec2.InstanceTypeInfo deallocated []*ec2.InstanceTypeInfo canAllocFn func(*ec2.InstanceTypeInfo, int) int + + reservationAllocErr error + reservationAllocated []*ec2.InstanceTypeInfo + reservationReleased []*ec2.InstanceTypeInfo + reservationAvailFn func(string, string, *ec2.InstanceTypeInfo) int } func (f *fakeResourceCapacityProvider) GetAvailableInstanceTypeInfos(showCapacity bool) []*ec2.InstanceTypeInfo { @@ -1007,6 +1012,25 @@ func (f *fakeResourceCapacityProvider) CanAllocate(it *ec2.InstanceTypeInfo, cou return count } +func (f *fakeResourceCapacityProvider) AllocateFromReservation(_, _ string, it *ec2.InstanceTypeInfo) error { + if f.reservationAllocErr != nil { + return f.reservationAllocErr + } + f.reservationAllocated = append(f.reservationAllocated, it) + return nil +} + +func (f *fakeResourceCapacityProvider) ReleaseToReservation(_ string, it *ec2.InstanceTypeInfo) { + f.reservationReleased = append(f.reservationReleased, it) +} + +func (f *fakeResourceCapacityProvider) ReservationAvailable(reservationID, accountID string, it *ec2.InstanceTypeInfo) int { + if f.reservationAvailFn != nil { + return f.reservationAvailFn(reservationID, accountID, it) + } + return 0 +} + func (f *fakeResourceCapacityProvider) InstanceTypes() map[string]*ec2.InstanceTypeInfo { return f.instanceTypes } @@ -2375,14 +2399,14 @@ func defaultPrepareInstanceTypes() (map[string]*ec2.InstanceTypeInfo, *ec2.Insta func TestPrepareRunInstances_MissingAccountID(t *testing.T) { svc := &InstanceServiceImpl{} - _, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{InstanceType: aws.String("t3.micro")}, "") + _, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{InstanceType: aws.String("t3.micro")}, "", "") require.Error(t, err) assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) } func TestPrepareRunInstances_MissingInstanceType(t *testing.T) { svc := &InstanceServiceImpl{} - _, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{}, "acc") + _, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{}, "acc", "") require.Error(t, err) assert.Equal(t, awserrors.ErrorMissingParameter, err.Error()) } @@ -2391,7 +2415,7 @@ func TestPrepareRunInstances_InvalidInstanceType(t *testing.T) { svc := &InstanceServiceImpl{instanceTypes: map[string]*ec2.InstanceTypeInfo{}} _, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{ InstanceType: aws.String("unknown.type"), - }, "acc") + }, "acc", "") require.Error(t, err) assert.Equal(t, awserrors.ErrorInvalidInstanceType, err.Error()) } @@ -2401,7 +2425,7 @@ func TestPrepareRunInstances_MissingImageID(t *testing.T) { svc := &InstanceServiceImpl{instanceTypes: types} _, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{ InstanceType: aws.String("t3.micro"), - }, "acc") + }, "acc", "") require.Error(t, err) assert.Equal(t, awserrors.ErrorMissingParameter, err.Error()) } @@ -2412,7 +2436,7 @@ func TestPrepareRunInstances_NilAMILoader(t *testing.T) { _, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{ InstanceType: aws.String("t3.micro"), ImageId: aws.String("ami-1"), - }, "acc") + }, "acc", "") require.Error(t, err) assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) } @@ -2426,7 +2450,7 @@ func TestPrepareRunInstances_AMINotFound(t *testing.T) { _, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{ InstanceType: aws.String("t3.micro"), ImageId: aws.String("ami-missing"), - }, "acc") + }, "acc", "") require.Error(t, err) assert.Equal(t, awserrors.ErrorInvalidAMIIDNotFound, err.Error()) } @@ -2442,7 +2466,7 @@ func TestPrepareRunInstances_AMINotOwnedByCaller(t *testing.T) { _, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{ InstanceType: aws.String("t3.micro"), ImageId: aws.String("ami-other"), - }, "111122223333") + }, "111122223333", "") require.Error(t, err) assert.Equal(t, awserrors.ErrorInvalidAMIIDNotFound, err.Error()) } @@ -2462,7 +2486,7 @@ func TestPrepareRunInstances_KeyPairNotFound(t *testing.T) { KeyName: aws.String("nope"), MinCount: aws.Int64(1), MaxCount: aws.Int64(1), - }, "acc") + }, "acc", "") require.Error(t, err) assert.Equal(t, awserrors.ErrorInvalidKeyPairNotFound, err.Error()) } @@ -2486,7 +2510,7 @@ func TestPrepareRunInstances_InsufficientCapacity(t *testing.T) { ImageId: aws.String("ami-1"), MinCount: aws.Int64(1), MaxCount: aws.Int64(5), - }, "acc") + }, "acc", "") require.Error(t, err) assert.Equal(t, awserrors.ErrorInsufficientInstanceCapacity, err.Error()) } @@ -2510,7 +2534,7 @@ func TestPrepareRunInstances_HappyPathNoENI(t *testing.T) { ImageId: aws.String("ami-1"), MinCount: aws.Int64(2), MaxCount: aws.Int64(2), - }, "acc") + }, "acc", "") require.NoError(t, err) require.NotNil(t, reservation) require.Len(t, instances, 2) @@ -2522,6 +2546,124 @@ func TestPrepareRunInstances_HappyPathNoENI(t *testing.T) { } } +// A targeted launch consumes slots from the reservation (not the general pool) +// and stamps the reservation id onto every prepared VM so terminate can restore. +func TestPrepareRunInstances_ConsumesReservation(t *testing.T) { + types, _ := defaultPrepareInstanceTypes() + prov := &fakeResourceCapacityProvider{ + instanceTypes: types, + reservationAvailFn: func(_, _ string, _ *ec2.InstanceTypeInfo) int { return 3 }, + } + svc := &InstanceServiceImpl{ + config: &config.Config{}, + instanceTypes: types, + amiLoader: &fakeAMILoader{byID: map[string]viperblock.AMIMetadata{ + "ami-1": {ImageOwnerAlias: "acc"}, + }}, + resourceMgr: prov, + } + _, instances, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{ + InstanceType: aws.String("t3.micro"), + ImageId: aws.String("ami-1"), + MinCount: aws.Int64(2), + MaxCount: aws.Int64(2), + }, "acc", "cr-123") + require.NoError(t, err) + require.Len(t, instances, 2) + assert.Len(t, prov.reservationAllocated, 2, "slots consumed from the reservation") + assert.Empty(t, prov.allocated, "targeted launch must not touch general capacity") + for _, inst := range instances { + assert.Equal(t, "cr-123", inst.CapacityReservationId, "instance stamped with reservation id") + } +} + +// A targeted launch is capped at the reservation's free slots and never spills +// onto general capacity: MaxCount past Available yields exactly Available. +func TestPrepareRunInstances_ReservationCapsNoSpill(t *testing.T) { + types, _ := defaultPrepareInstanceTypes() + prov := &fakeResourceCapacityProvider{ + instanceTypes: types, + reservationAvailFn: func(_, _ string, _ *ec2.InstanceTypeInfo) int { return 2 }, + } + svc := &InstanceServiceImpl{ + config: &config.Config{}, + instanceTypes: types, + amiLoader: &fakeAMILoader{byID: map[string]viperblock.AMIMetadata{ + "ami-1": {ImageOwnerAlias: "acc"}, + }}, + resourceMgr: prov, + } + _, instances, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{ + InstanceType: aws.String("t3.micro"), + ImageId: aws.String("ami-1"), + MinCount: aws.Int64(1), + MaxCount: aws.Int64(5), + }, "acc", "cr-123") + require.NoError(t, err) + assert.Len(t, instances, 2, "launch capped at the 2 available reservation slots") + assert.Empty(t, prov.allocated, "no general-pool allocation for a targeted launch") +} + +// When the reservation has fewer free slots than MinCount the launch is rejected +// with ReservationCapacityExceeded and nothing is allocated. +func TestPrepareRunInstances_ReservationExceeded(t *testing.T) { + types, _ := defaultPrepareInstanceTypes() + prov := &fakeResourceCapacityProvider{ + instanceTypes: types, + reservationAvailFn: func(_, _ string, _ *ec2.InstanceTypeInfo) int { return 1 }, + } + svc := &InstanceServiceImpl{ + config: &config.Config{}, + instanceTypes: types, + amiLoader: &fakeAMILoader{byID: map[string]viperblock.AMIMetadata{ + "ami-1": {ImageOwnerAlias: "acc"}, + }}, + resourceMgr: prov, + } + _, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{ + InstanceType: aws.String("t3.micro"), + ImageId: aws.String("ami-1"), + MinCount: aws.Int64(2), + MaxCount: aws.Int64(2), + }, "acc", "cr-123") + require.Error(t, err) + assert.Equal(t, awserrors.ErrorReservationCapacityExceeded, err.Error()) + assert.Empty(t, prov.reservationAllocated, "nothing allocated when below MinCount") +} + +// Invariant guardrail: a mid-launch failure on the reservation path must return +// every consumed slot to the reservation (reservationReleased == reservationAllocated) +// and never leak one onto the general pool (Deallocate untouched). Each per-instance +// ENI create fails, exercising a rollback site inside the launch loop. +func TestPrepareRunInstances_ReservationRollbackNoGeneralPoolLeak(t *testing.T) { + types, _ := defaultPrepareInstanceTypes() + prov := &fakeResourceCapacityProvider{ + instanceTypes: types, + reservationAvailFn: func(_, _ string, _ *ec2.InstanceTypeInfo) int { return 2 }, + } + eni := &fakeENICreator{createErr: errors.New("eni create failed")} // fails every call + svc := &InstanceServiceImpl{ + config: &config.Config{Region: "us-east-1", AZ: "us-east-1a"}, + instanceTypes: types, + amiLoader: &fakeAMILoader{byID: map[string]viperblock.AMIMetadata{ + "ami-1": {ImageOwnerAlias: "acc"}, + }}, + resourceMgr: prov, + eniCreator: eni, + } + _, _, _, err := svc.PrepareRunInstances(&ec2.RunInstancesInput{ + InstanceType: aws.String("t3.micro"), + ImageId: aws.String("ami-1"), + SubnetId: aws.String("subnet-1"), + MinCount: aws.Int64(2), + MaxCount: aws.Int64(2), + }, "acc", "cr-123") + require.Error(t, err) + assert.Len(t, prov.reservationAllocated, 2, "both slots allocated from the reservation") + assert.Len(t, prov.reservationReleased, 2, "both reservation slots restored on rollback") + assert.Empty(t, prov.deallocated, "no reservation-bound slot may leak to the general pool") +} + // TestPrepareRunInstances_AmiLaunchIndexContiguous pins that ami-launch-index is // assigned per successful launch (0..n-1) so it flows to DescribeInstances and the // IMDS identity document. Survivors of a mid-loop failure stay contiguous (no gap). @@ -2545,7 +2687,7 @@ func TestPrepareRunInstances_AmiLaunchIndexContiguous(t *testing.T) { ImageId: aws.String("ami-1"), MinCount: aws.Int64(3), MaxCount: aws.Int64(3), - }, "acc") + }, "acc", "") require.NoError(t, err) require.Len(t, instances, 3) require.Len(t, reservation.Instances, 3) @@ -2578,7 +2720,7 @@ func TestPrepareRunInstances_AmiLaunchIndexContiguous(t *testing.T) { SubnetId: aws.String("subnet-1"), MinCount: aws.Int64(2), MaxCount: aws.Int64(3), - }, "acc") + }, "acc", "") require.NoError(t, err) require.Len(t, instances, 2) require.Len(t, reservation.Instances, 2) @@ -2621,7 +2763,7 @@ func TestPrepareRunInstances_BootModePropagated(t *testing.T) { ImageId: aws.String("ami-1"), MinCount: aws.Int64(1), MaxCount: aws.Int64(1), - }, "acc") + }, "acc", "") require.NoError(t, err) require.Len(t, instances, 1) assert.Equal(t, tc.wantBootMode, instances[0].BootMode) @@ -2826,7 +2968,7 @@ func TestPrepareRunInstances_DefaultSubnetResolved(t *testing.T) { ImageId: aws.String("ami-1"), MinCount: aws.Int64(1), MaxCount: aws.Int64(1), - }, "acc") + }, "acc", "") require.NoError(t, err) require.Len(t, instances, 1) assert.Equal(t, "eni-1", instances[0].ENIId) @@ -2882,7 +3024,7 @@ func TestPrepareRunInstances_PublicIPAutoAssigned(t *testing.T) { } } - _, instances, _, err := svc.PrepareRunInstances(input, "acc") + _, instances, _, err := svc.PrepareRunInstances(input, "acc", "") require.NoError(t, err) require.Len(t, instances, 1) if tc.wantPublic { @@ -2949,7 +3091,7 @@ func TestPrepareRunInstances_NATFailureRollsBackPublicIP(t *testing.T) { SubnetId: aws.String("subnet-1"), MinCount: aws.Int64(1), MaxCount: aws.Int64(1), - }, "acc") + }, "acc", "") // MinCount=1 with a single launch attempt that NAT-failed must drop // below minimum and return ServerInternal — the wrapped natErr is not @@ -3018,7 +3160,7 @@ func TestPrepareRunInstances_PublicIPAllocFailureAbortsLaunch(t *testing.T) { SubnetId: aws.String("subnet-1"), MinCount: aws.Int64(1), MaxCount: aws.Int64(1), - }, "acc") + }, "acc", "") require.Error(t, err) assert.Equal(t, awserrors.ErrorInsufficientAddressCapacity, err.Error(), @@ -3053,7 +3195,7 @@ func TestPrepareRunInstances_ENICreateFailureDeallocates(t *testing.T) { SubnetId: aws.String("subnet-1"), MinCount: aws.Int64(1), MaxCount: aws.Int64(1), - }, "acc") + }, "acc", "") require.Error(t, err) // MinCount not satisfied → InsufficientInstanceCapacity (no known // AWS code for raw ENI failure). @@ -3086,7 +3228,7 @@ func TestPrepareRunInstances_ENIAttachFailureRollsBack(t *testing.T) { SubnetId: aws.String("subnet-1"), MinCount: aws.Int64(1), MaxCount: aws.Int64(1), - }, "acc") + }, "acc", "") require.Error(t, err) assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) @@ -3122,7 +3264,7 @@ func TestPrepareRunInstances_NetworkInterfaceLifted(t *testing.T) { Groups: []*string{aws.String("sg-1")}, }, }, - }, "acc") + }, "acc", "") require.NoError(t, err) require.Len(t, instances, 1) assert.Equal(t, "eni-3", instances[0].ENIId) @@ -3148,7 +3290,7 @@ func TestPrepareRunInstances_PlacementGroup(t *testing.T) { MinCount: aws.Int64(1), MaxCount: aws.Int64(1), Placement: &ec2.Placement{GroupName: aws.String("pg-1")}, - }, "acc") + }, "acc", "") require.NoError(t, err) require.Len(t, instances, 1) assert.Equal(t, "pg-1", instances[0].PlacementGroupName) @@ -3174,7 +3316,7 @@ func TestPrepareRunInstances_AllocateFailsMidLoop(t *testing.T) { ImageId: aws.String("ami-1"), MinCount: aws.Int64(1), MaxCount: aws.Int64(3), - }, "acc") + }, "acc", "") require.Error(t, err) assert.Equal(t, awserrors.ErrorInsufficientInstanceCapacity, err.Error()) } @@ -3530,7 +3672,7 @@ func TestPrepareRunInstances_PreCreatedENIAttachedSkipsAutoCreate(t *testing.T) NetworkInterfaceId: aws.String("eni-pre"), DeviceIndex: aws.Int64(0), }}, - }, "acc") + }, "acc", "") require.NoError(t, err) require.Len(t, instances, 1) assert.Equal(t, "eni-pre", instances[0].ENIId) @@ -3558,7 +3700,7 @@ func TestPrepareRunInstances_PreCreatedENIInUseRejected(t *testing.T) { NetworkInterfaces: []*ec2.InstanceNetworkInterfaceSpecification{{ NetworkInterfaceId: aws.String("eni-busy"), }}, - }, "acc") + }, "acc", "") require.Error(t, err) assert.Equal(t, awserrors.ErrorInvalidNetworkInterfaceInUse, err.Error()) assert.Equal(t, 0, eni.attachCalls, "in-use ENI must not be attached") @@ -3579,7 +3721,7 @@ func TestPrepareRunInstances_PreCreatedENILookupErrorSurfaced(t *testing.T) { NetworkInterfaces: []*ec2.InstanceNetworkInterfaceSpecification{{ NetworkInterfaceId: aws.String("eni-ghost"), }}, - }, "acc") + }, "acc", "") require.Error(t, err) assert.Equal(t, awserrors.ErrorInvalidNetworkInterfaceIDNotFound, err.Error()) assert.Equal(t, 0, eni.attachCalls) @@ -3610,7 +3752,7 @@ func TestPrepareRunInstances_PreCreatedENIAttachErrorSurfaced(t *testing.T) { NetworkInterfaces: []*ec2.InstanceNetworkInterfaceSpecification{{ NetworkInterfaceId: aws.String("eni-pre"), }}, - }, "acc") + }, "acc", "") require.Error(t, err) assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) assert.Equal(t, 1, eni.attachCalls) diff --git a/spinifex/vm/crash_recovery.go b/spinifex/vm/crash_recovery.go index eb1b34323..075d5d282 100644 --- a/spinifex/vm/crash_recovery.go +++ b/spinifex/vm/crash_recovery.go @@ -114,7 +114,16 @@ func (m *Manager) HandleCrash(instance *VM, waitErr error) { if m.deps.Resources != nil && instance.InstanceType != "" { slog.Info("Deallocating resources for crashed instance", "instance", instance.ID, "type", instance.InstanceType) - m.deps.Resources.Deallocate(instance.InstanceType) + m.deallocateResources(instance) + } + + // The slot just returned to the reservation (deallocateResources above), but + // the restart re-allocates from the general pool, so drop the binding under the + // manager lock. This keeps the crash release and restart allocation on the same + // pool; leaving it set would drift the shared reservation counter on the later + // stop/terminate while other instances still consume the reservation. + if instance.CapacityReservationId != "" { + m.UpdateState(instance.ID, func(v *VM) { v.CapacityReservationId = "" }) } if instance.Config.QMPSocket != "" { @@ -246,7 +255,7 @@ func (m *Manager) RestartCrashedInstance(instance *VM) { if err := m.deps.TransitionState(instance, StatePending); err != nil { slog.Error("Failed to transition instance to pending for restart", "instance", instance.ID, "err", err) - m.deps.Resources.Deallocate(instance.InstanceType) + m.deallocateResources(instance) return } } diff --git a/spinifex/vm/crash_recovery_test.go b/spinifex/vm/crash_recovery_test.go index 96a02b042..76432f716 100644 --- a/spinifex/vm/crash_recovery_test.go +++ b/spinifex/vm/crash_recovery_test.go @@ -161,6 +161,53 @@ func TestHandleCrash_CoreFlow(t *testing.T) { assert.True(t, os.IsNotExist(err), "QMP socket must be removed") } +// A reservation-bound instance returns its slot to the reservation through the +// deallocateResources chokepoint, never to the general pool (net-zero swap). +func TestHandleCrash_ReservationBoundReleasesToReservation(t *testing.T) { + m, rc, _, _ := crashTestManager(t) + + tmpDir := t.TempDir() + qmpPath := filepath.Join(tmpDir, "qmp.sock") + require.NoError(t, os.WriteFile(qmpPath, []byte("dummy"), 0o600)) + + instance := &VM{ + ID: "i-cr", + Status: StateRunning, + InstanceType: "t3.micro", + CapacityReservationId: "cr-0123456789abcdef0", + Config: Config{QMPSocket: qmpPath}, + } + m.Insert(instance) + + m.HandleCrash(instance, fmt.Errorf("test crash")) + + assert.Equal(t, 1, rc.releaseCount(), "slot must return to the reservation") + assert.Zero(t, rc.deallocateCount("t3.micro"), "reservation-bound release must not hit the general pool") +} + +// A crashed reservation-bound instance drops its binding: the slot went back to +// the reservation, and the restart runs on general capacity, so a later +// stop/terminate must not double-credit the reservation's shared counter. +func TestHandleCrash_ReservationBoundClearsBindingForRestart(t *testing.T) { + m, rc, _, _ := crashTestManager(t) + + instance := &VM{ + ID: "i-cr-clear", + Status: StateRunning, + InstanceType: "t3.micro", + CapacityReservationId: "cr-0123456789abcdef0", + } + m.Insert(instance) + + m.HandleCrash(instance, fmt.Errorf("test crash")) + + assert.Equal(t, 1, rc.releaseCount(), "slot must return to the reservation on crash") + assert.Empty(t, instance.CapacityReservationId, "binding must be cleared so restart uses general capacity") + stored, ok := m.Get(instance.ID) + require.True(t, ok) + assert.Empty(t, stored.CapacityReservationId, "cleared binding must be persisted on the stored VM") +} + func TestHandleCrash_FirstCrashSetsTime(t *testing.T) { m, _, _, _ := crashTestManager(t) diff --git a/spinifex/vm/fakes_test.go b/spinifex/vm/fakes_test.go index 1ca198c1d..d50a46f55 100644 --- a/spinifex/vm/fakes_test.go +++ b/spinifex/vm/fakes_test.go @@ -9,6 +9,7 @@ type fakeResourceController struct { mu sync.Mutex allocated map[string]int deallocated []string + released []string // "/" per ReleaseToReservation call allocateErr error canAllocateRet int } @@ -36,6 +37,21 @@ func (f *fakeResourceController) Deallocate(typeName string) { } } +func (f *fakeResourceController) ReleaseToReservation(reservationID, typeName string) { + f.mu.Lock() + defer f.mu.Unlock() + f.released = append(f.released, reservationID+"/"+typeName) + if f.allocated[typeName] > 0 { + f.allocated[typeName]-- + } +} + +func (f *fakeResourceController) releaseCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.released) +} + func (f *fakeResourceController) CanAllocate(typeName string, count int) int { f.mu.Lock() defer f.mu.Unlock() diff --git a/spinifex/vm/instance_type_resolver.go b/spinifex/vm/instance_type_resolver.go index 379b262ed..402d65740 100644 --- a/spinifex/vm/instance_type_resolver.go +++ b/spinifex/vm/instance_type_resolver.go @@ -21,6 +21,11 @@ type InstanceTypeResolver interface { type ResourceController interface { Allocate(instanceType string) error Deallocate(instanceType string) + // ReleaseToReservation returns instanceType's compute to the capacity + // reservation reservationID instead of the general pool — the net-zero + // inverse of the launch-time swap, used when a reservation-bound instance + // stops or terminates. Frees to the general pool if the reservation is gone. + ReleaseToReservation(reservationID, instanceType string) // CanAllocate returns how many instances of instanceType could be // allocated right now (0 to count). Used by the crash-recovery // scheduler to avoid restarting a VM the host can't fit. diff --git a/spinifex/vm/shutdown.go b/spinifex/vm/shutdown.go index af5271caf..303b12deb 100644 --- a/spinifex/vm/shutdown.go +++ b/spinifex/vm/shutdown.go @@ -282,6 +282,14 @@ func (m *Manager) stopCleanup(instance *VM) { } } m.deallocateResources(instance) + + // The slot just returned to the reservation; detach the binding (under the + // manager lock, mirroring crash recovery) so a later start re-allocates from + // the general pool and terminate frees there too — never double-counting the + // reservation while the stopped instance no longer holds a slot. + if instance.CapacityReservationId != "" { + m.UpdateState(instance.ID, func(v *VM) { v.CapacityReservationId = "" }) + } } // terminateCleanup is stopCleanup plus the AWS-resource cleanup that @@ -398,11 +406,20 @@ func (m *Manager) cleanupExtraENITaps(instance *VM) { } // deallocateResources releases the per-instance vCPU/memory reservation -// back to the resource controller. +// back to the resource controller. The single release/restore chokepoint for +// stop, terminate and crash recovery. func (m *Manager) deallocateResources(instance *VM) { if m.deps.Resources == nil || instance.InstanceType == "" { return } + // A reservation-bound instance returns its slot to the reservation, not the + // general pool. CapacityReservationId is set at launch and only cleared (under + // the manager lock, on crash before a general-capacity restart), so the + // stop/terminate/crash reads here do not overlap that clear. + if instance.CapacityReservationId != "" { + m.deps.Resources.ReleaseToReservation(instance.CapacityReservationId, instance.InstanceType) + return + } m.deps.Resources.Deallocate(instance.InstanceType) } diff --git a/spinifex/vm/shutdown_test.go b/spinifex/vm/shutdown_test.go index eacacd865..02df3f31b 100644 --- a/spinifex/vm/shutdown_test.go +++ b/spinifex/vm/shutdown_test.go @@ -555,6 +555,30 @@ func TestStopCleanup_InvokesReleaseGPU(t *testing.T) { } } +// TestStopCleanup_ReservationBoundReleasesAndClears proves a reservation-bound +// instance returns its slot to the reservation on stop and then detaches the +// binding, so a later start re-allocates from the general pool and terminate +// frees there too — never double-counting the reservation. +func TestStopCleanup_ReservationBoundReleasesAndClears(t *testing.T) { + t.Setenv("XDG_RUNTIME_DIR", t.TempDir()) + rc := &countingResourceController{} + m := NewManagerWithDeps(Deps{Resources: rc}) + instance := &VM{ID: "i-cr", InstanceType: "t3.micro", CapacityReservationId: "cr-123"} + m.Insert(instance) + + m.stopCleanup(instance) + + if got := rc.releasedToCRID; len(got) != 1 || got[0] != "cr-123" { + t.Fatalf("stop must release the slot to the reservation: got %v, want [cr-123]", got) + } + if rc.deallocations != 0 { + t.Fatalf("stop must not free a reservation-bound instance to the general pool: deallocations=%d", rc.deallocations) + } + if instance.CapacityReservationId != "" { + t.Fatalf("stop must clear the reservation binding so start uses general capacity: got %q", instance.CapacityReservationId) + } +} + func TestTerminateCleanup_InvokesReleaseGPU(t *testing.T) { t.Setenv("XDG_RUNTIME_DIR", t.TempDir()) cleaner := &recordingInstanceCleaner{} @@ -1082,13 +1106,20 @@ func TestClassifyRestoredInstances_StateErrorSkipsRelaunch(t *testing.T) { } // countingResourceController counts how often Allocate fires so a test -// can prove the restore path skipped its resource-allocation branch. +// can prove the restore path skipped its resource-allocation branch. It +// also records general deallocations and reservation releases so stop +// tests can assert which pool a reservation-bound instance returns to. type countingResourceController struct { - allocations int + allocations int + deallocations int + releasedToCRID []string } -func (c *countingResourceController) Allocate(_ string) error { c.allocations++; return nil } -func (c *countingResourceController) Deallocate(_ string) {} +func (c *countingResourceController) Allocate(_ string) error { c.allocations++; return nil } +func (c *countingResourceController) Deallocate(_ string) { c.deallocations++ } +func (c *countingResourceController) ReleaseToReservation(crID, _ string) { + c.releasedToCRID = append(c.releasedToCRID, crID) +} func (c *countingResourceController) CanAllocate(_ string, n int) int { return n } // signalingStore is a no-op StateStore that fires onSave once per diff --git a/spinifex/vm/vm.go b/spinifex/vm/vm.go index e8670d89c..ac1ccb1ac 100644 --- a/spinifex/vm/vm.go +++ b/spinifex/vm/vm.go @@ -140,6 +140,11 @@ type VM struct { // the visibility window, after which it may reclaim it early; the bucket's // 1h TTL bounds visibility regardless. TerminatedAt time.Time `json:"terminated_at,omitzero"` + + // CapacityReservationId is the On-Demand Capacity Reservation this instance + // was launched into (targeted launch). Empty for instances on general + // capacity. Drives slot restore to the reservation on stop/terminate. + CapacityReservationId string `json:"capacity_reservation_id,omitempty"` } // ResetNodeLocalState zeroes node-specific fields after deserializing a VM