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/docs/service-interfaces.yaml b/docs/service-interfaces.yaml index eea6c10cb..a92305ec8 100644 --- a/docs/service-interfaces.yaml +++ b/docs/service-interfaces.yaml @@ -292,6 +292,14 @@ suites: # via vpcd, AMI via predastore, root volume via viperblockd, lifecycle via # spinifex_daemon. Single-mode, VM-booting leg. + e2e-ecr: + path: tests/e2e/ecr + covers: [awsgw, spinifex_daemon, predastore] + # ECR control plane (handlers/ecr, gateway/ecr*) + OCI data plane: blobs and + # manifests in predastore, control + auth bridge in awsgw, KV via + # spinifex_daemon. docker/crane/skopeo data-plane subtests skip when the + # client binary or registry-host resolution is absent. + e2e-ddil: path: tests/e2e/ddil covers: [awsgw, spinifex_daemon, vpcd, viperblockd, predastore] 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/gateway/ecr/registry.go b/spinifex/gateway/ecr/registry.go index 6c7cf1363..9bafc6092 100644 --- a/spinifex/gateway/ecr/registry.go +++ b/spinifex/gateway/ecr/registry.go @@ -516,7 +516,7 @@ func (reg *Registry) getManifest(w http.ResponseWriter, r *http.Request, name, r WriteError(w, http.StatusNotFound, "MANIFEST_UNKNOWN", "manifest unknown") return } - if !acceptsType(r.Header.Get("Accept"), meta.MediaType) { + if !acceptsType(strings.Join(r.Header.Values("Accept"), ","), meta.MediaType) { WriteError(w, http.StatusNotAcceptable, "MANIFEST_INVALID", "no acceptable manifest media type") return } diff --git a/spinifex/gateway/ecr/registry_test.go b/spinifex/gateway/ecr/registry_test.go index c716cd623..b0ccef3e0 100644 --- a/spinifex/gateway/ecr/registry_test.go +++ b/spinifex/gateway/ecr/registry_test.go @@ -208,6 +208,29 @@ func TestManifest_Get_AcceptMismatch(t *testing.T) { assert.Equal(t, http.StatusNotAcceptable, w.Code) } +// TestManifest_Get_MultipleAcceptHeaders reproduces skopeo, which sends each +// acceptable manifest media type as its own Accept header line rather than one +// comma-joined value. Negotiation must consider every line, not just the first. +func TestManifest_Get_MultipleAcceptHeaders(t *testing.T) { + reg := newTestRegistry() + repo := "team/app" + cfg := pushBlob(t, reg, repo, []byte("c")) + manifest := fmt.Appendf(nil, + `{"mediaType":"%s","config":{"digest":"%s"},"layers":[]}`, mediaTypeDockerManifest, cfg) + do(reg, http.MethodPut, "/v2/"+repo+"/manifests/v1", manifest, + map[string]string{"Content-Type": mediaTypeDockerManifest}) + + r := httptest.NewRequest(http.MethodGet, "/v2/"+repo+"/manifests/v1", nil) + r.Header.Add("Accept", mediaTypeOCIIndex) + r.Header.Add("Accept", mediaTypeOCIManifest) + r.Header.Add("Accept", mediaTypeDockerManifest) + w := httptest.NewRecorder() + reg.ServeHTTP(w, r) + + assert.Equal(t, http.StatusOK, w.Code, "body: %s", w.Body.String()) + assert.Equal(t, manifest, w.Body.Bytes()) +} + func TestManifest_Get_Unknown(t *testing.T) { reg := newTestRegistry() w := do(reg, http.MethodGet, "/v2/team/app/manifests/missing", nil, nil) 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 diff --git a/tests/e2e/Makefile b/tests/e2e/Makefile index 055336de5..cd5ee24c4 100644 --- a/tests/e2e/Makefile +++ b/tests/e2e/Makefile @@ -1,4 +1,4 @@ -PKG ?= ./cert/... ./lb/... ./eks/... ./iam/... ./multinode/... ./baremetal/... ./reboot/... ./tofu/... +PKG ?= ./cert/... ./lb/... ./eks/... ./iam/... ./ecr/... ./multinode/... ./baremetal/... ./reboot/... ./tofu/... ARTIFACT_DIR ?= /tmp/spinifex-e2e-artifacts TIMEOUT ?= 30m MODE ?= single @@ -11,9 +11,9 @@ SUITE_TAG ?= $(notdir $(patsubst %/...,%,$(firstword $(PKG)))) # named suite into BIN_DIR/.test so CI can ship binaries (not source + # go toolchain) to the remote VM. BIN_DIR ?= $(CURDIR)/_bin -SUITES_BUILD ?= cert lb eks single iam gpu +SUITES_BUILD ?= cert lb eks single iam ecr gpu -.PHONY: e2e e2e-cert e2e-lb e2e-eks e2e-single e2e-iam e2e-multinode e2e-bm e2e-reboot e2e-tofu e2e-gpu e2e-all clean e2e-build +.PHONY: e2e e2e-cert e2e-lb e2e-eks e2e-single e2e-iam e2e-ecr e2e-multinode e2e-bm e2e-reboot e2e-tofu e2e-gpu e2e-all clean e2e-build ifeq ($(GOTESTSUM),1) E2E_CMD = gotestsum \ @@ -37,6 +37,7 @@ e2e-lb: ; $(MAKE) e2e PKG=./lb/... e2e-eks: ; $(MAKE) e2e PKG=./eks/... e2e-single: ; $(MAKE) e2e PKG=./single/... e2e-iam: ; $(MAKE) e2e PKG=./iam/... +e2e-ecr: ; $(MAKE) e2e PKG=./ecr/... e2e-multinode: ; $(MAKE) e2e PKG=./multinode/... MODE=multinode e2e-bm: ; $(MAKE) e2e PKG=./baremetal/... MODE=baremetal e2e-reboot: ; $(MAKE) e2e PKG=./reboot/... diff --git a/tests/e2e/ecr/capacity_test.go b/tests/e2e/ecr/capacity_test.go new file mode 100644 index 000000000..b63a0b46b --- /dev/null +++ b/tests/e2e/ecr/capacity_test.go @@ -0,0 +1,14 @@ +//go:build e2e + +package ecr + +import "testing" + +// TestECRCapacityLoad is the predastore capacity/load leg of Sprint 2f (parallel +// crane push of many large images, asserting predastore throughput with no +// manifest corruption). Deferred: it depends on predastore fixes mulga-siv-359 +// (':'-in-key SigV4 canonicalization) and mulga-siv-356 (404 NoSuchBucket), +// which are owned elsewhere. Unskip once those land. +func TestECRCapacityLoad(t *testing.T) { + t.Skip("deferred: blocked on predastore mulga-siv-359 / mulga-siv-356") +} diff --git a/tests/e2e/ecr/crane_skopeo_test.go b/tests/e2e/ecr/crane_skopeo_test.go new file mode 100644 index 000000000..3eb5fb40a --- /dev/null +++ b/tests/e2e/ecr/crane_skopeo_test.go @@ -0,0 +1,114 @@ +//go:build e2e + +package ecr + +import ( + "archive/tar" + "crypto/rand" + "os" + "path/filepath" + "testing" + "time" + + "github.com/mulgadc/spinifex/tests/e2e/harness" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ociCmdTimeout = 2 * time.Minute + +// writeLayerTar writes a single-file tar (an OCI layer) with random contents and +// returns its path. +func writeLayerTar(t *testing.T, dir string) string { + t.Helper() + path := filepath.Join(dir, "layer.tar") + fh, err := os.Create(path) //nolint:gosec // test-controlled path under TmpDir + require.NoError(t, err) + defer fh.Close() + + payload := make([]byte, 8192) + _, err = rand.Read(payload) + require.NoError(t, err) + + tw := tar.NewWriter(fh) + require.NoError(t, tw.WriteHeader(&tar.Header{ + Name: "payload", Mode: 0o600, Size: int64(len(payload)), + })) + _, err = tw.Write(payload) + require.NoError(t, err) + require.NoError(t, tw.Close()) + return path +} + +// craneLogin authenticates crane against host, writing the auth file under +// dockerConfig so subsequent crane calls reuse it. +func craneLogin(t *testing.T, crane *harness.ExternalCLI, host, pass string) { + t.Helper() + out, err := crane.Run(ociCmdTimeout, "auth", "login", host, "-u", "AWS", "-p", pass) + require.NoErrorf(t, err, "crane auth login: %s", out) +} + +// TestCranePushPull pushes a scratch image with `crane append` and pulls it +// back, exercising the Bearer /v2/token flow crane follows. +func TestCranePushPull(t *testing.T) { + f := requireECRFixture(t) + host := harness.ECRRegistryHost(f.Account) + harness.RequireRegistryResolves(t, host) + + dockerConfig := filepath.Join(f.TmpDir, "crane") + require.NoError(t, os.MkdirAll(dockerConfig, 0o700)) + crane := harness.NewCrane(t, dockerConfig) + craneLogin(t, crane, host, harness.ECRGetLoginPassword(t, f.AWS)) + + repo := uniqueRepo("crane") + harness.CreateECRRepository(t, f.AWS, repo) + ref := harness.ECRRepositoryURI(f.Account, repo) + ":v1" + + layer := writeLayerTar(t, f.TmpDir) + out, err := crane.Run(ociCmdTimeout, "append", "-f", layer, "-t", ref) + require.NoErrorf(t, err, "crane append: %s", out) + + require.True(t, harness.ECRWaitImageTag(t, f.AWS, repo, "v1", 30*time.Second), + "crane-pushed tag v1 not visible via DescribeImages") + + digest, err := crane.Run(ociCmdTimeout, "digest", ref) + require.NoErrorf(t, err, "crane digest: %s", digest) + assert.Contains(t, digest, "sha256:") + + out, err = crane.Run(ociCmdTimeout, "pull", ref, filepath.Join(f.TmpDir, "pulled.tar")) + require.NoErrorf(t, err, "crane pull: %s", out) +} + +// TestSkopeoCopy seeds an image with crane, then `skopeo copy` retags it within +// the registry — exercising skopeo's Bearer auth on both read and write. +func TestSkopeoCopy(t *testing.T) { + f := requireECRFixture(t) + host := harness.ECRRegistryHost(f.Account) + harness.RequireRegistryResolves(t, host) + + dockerConfig := filepath.Join(f.TmpDir, "skopeo") + require.NoError(t, os.MkdirAll(dockerConfig, 0o700)) + crane := harness.NewCrane(t, dockerConfig) + skopeo := harness.NewSkopeo(t, dockerConfig) + + pass := harness.ECRGetLoginPassword(t, f.AWS) + craneLogin(t, crane, host, pass) + + repo := uniqueRepo("skopeo") + harness.CreateECRRepository(t, f.AWS, repo) + srcRef := harness.ECRRepositoryURI(f.Account, repo) + ":src" + dstRef := harness.ECRRepositoryURI(f.Account, repo) + ":dst" + + layer := writeLayerTar(t, f.TmpDir) + out, err := crane.Run(ociCmdTimeout, "append", "-f", layer, "-t", srcRef) + require.NoErrorf(t, err, "crane append: %s", out) + + creds := "AWS:" + pass + out, err = skopeo.Run(ociCmdTimeout, "copy", + "--src-creds", creds, "--dest-creds", creds, + "docker://"+srcRef, "docker://"+dstRef) + require.NoErrorf(t, err, "skopeo copy: %s", out) + + require.True(t, harness.ECRWaitImageTag(t, f.AWS, repo, "dst", 30*time.Second), + "skopeo-copied tag dst not visible via DescribeImages") +} diff --git a/tests/e2e/ecr/docker_push_test.go b/tests/e2e/ecr/docker_push_test.go new file mode 100644 index 000000000..166678b1c --- /dev/null +++ b/tests/e2e/ecr/docker_push_test.go @@ -0,0 +1,80 @@ +//go:build e2e + +package ecr + +import ( + "crypto/rand" + "os" + "path/filepath" + "testing" + "time" + + "github.com/mulgadc/spinifex/tests/e2e/harness" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const dockerCmdTimeout = 2 * time.Minute + +// buildScratchContext writes a hermetic FROM-scratch build context (no network +// pulls) with a random payload layer, returning the context dir. +func buildScratchContext(t *testing.T, dir string) string { + t.Helper() + payload := make([]byte, 4096) + _, err := rand.Read(payload) + require.NoError(t, err) + require.NoError(t, os.WriteFile(filepath.Join(dir, "payload"), payload, 0o600)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "Dockerfile"), + []byte("FROM scratch\nCOPY payload /payload\n"), 0o600)) + return dir +} + +// TestDockerPushPull drives the canonical workflow: aws ecr get-login-password +// -> docker login -> docker push -> DescribeImages -> docker pull (clean daemon) +// against the cluster registry. +func TestDockerPushPull(t *testing.T) { + f := requireECRFixture(t) + host := harness.ECRRegistryHost(f.Account) + harness.RequireRegistryResolves(t, host) + + dockerConfig := filepath.Join(f.TmpDir, "docker") + require.NoError(t, os.MkdirAll(dockerConfig, 0o700)) + docker := harness.NewDocker(t, dockerConfig) + + repo := uniqueRepo("docker") + harness.CreateECRRepository(t, f.AWS, repo) + ref := harness.ECRRepositoryURI(f.Account, repo) + ":v1" + + t.Run("docker login", func(t *testing.T) { + pass := harness.ECRGetLoginPassword(t, f.AWS) + out, err := docker.RunStdin(dockerCmdTimeout, pass, + "login", host, "-u", "AWS", "--password-stdin") + require.NoErrorf(t, err, "docker login: %s", out) + assert.Contains(t, out, "Login Succeeded") + }) + + t.Run("docker build + push", func(t *testing.T) { + buildDir := filepath.Join(f.TmpDir, "build") + require.NoError(t, os.MkdirAll(buildDir, 0o700)) + ctxDir := buildScratchContext(t, buildDir) + + out, err := docker.Run(dockerCmdTimeout, "build", "-t", ref, ctxDir) + require.NoErrorf(t, err, "docker build: %s", out) + + out, err = docker.Run(dockerCmdTimeout, "push", ref) + require.NoErrorf(t, err, "docker push: %s", out) + }) + + t.Run("DescribeImages shows the pushed tag", func(t *testing.T) { + require.True(t, harness.ECRWaitImageTag(t, f.AWS, repo, "v1", 30*time.Second), + "pushed tag v1 not visible via DescribeImages") + }) + + t.Run("docker pull round-trips from a clean daemon", func(t *testing.T) { + out, err := docker.Run(dockerCmdTimeout, "rmi", "-f", ref) + require.NoErrorf(t, err, "docker rmi: %s", out) + + out, err = docker.Run(dockerCmdTimeout, "pull", ref) + require.NoErrorf(t, err, "docker pull: %s", out) + }) +} diff --git a/tests/e2e/ecr/ecr_test.go b/tests/e2e/ecr/ecr_test.go new file mode 100644 index 000000000..d0081aa29 --- /dev/null +++ b/tests/e2e/ecr/ecr_test.go @@ -0,0 +1,106 @@ +//go:build e2e + +package ecr + +import ( + "fmt" + "time" + + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ecr" + "github.com/mulgadc/spinifex/tests/e2e/harness" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// uniqueRepo returns a collision-resistant repo name scoped to a test. +func uniqueRepo(prefix string) string { + return fmt.Sprintf("e2e/%s-%d", prefix, time.Now().UnixNano()) +} + +// TestECRControlPlane round-trips the ECR control-plane surface: repository +// CRUD, repository policy, and lifecycle policy. Zero VMs, pure SDK. +func TestECRControlPlane(t *testing.T) { + f := requireECRFixture(t) + c := f.AWS + repo := uniqueRepo("ctl") + harness.CreateECRRepository(t, c, repo) + + t.Run("DescribeRepositories lists the new repo", func(t *testing.T) { + out, err := c.ECR.DescribeRepositories(&ecr.DescribeRepositoriesInput{ + RepositoryNames: []*string{aws.String(repo)}, + }) + require.NoError(t, err) + require.Len(t, out.Repositories, 1) + assert.Equal(t, repo, aws.StringValue(out.Repositories[0].RepositoryName)) + assert.Contains(t, aws.StringValue(out.Repositories[0].RepositoryUri), repo) + }) + + t.Run("repository policy round-trips", func(t *testing.T) { + policy := `{"Version":"2012-10-17","Statement":[{"Sid":"allow-pull","Effect":"Allow","Principal":"*","Action":["ecr:GetDownloadUrlForLayer","ecr:BatchGetImage"]}]}` + _, err := c.ECR.SetRepositoryPolicy(&ecr.SetRepositoryPolicyInput{ + RepositoryName: aws.String(repo), + PolicyText: aws.String(policy), + }) + require.NoError(t, err) + + got, err := c.ECR.GetRepositoryPolicy(&ecr.GetRepositoryPolicyInput{ + RepositoryName: aws.String(repo), + }) + require.NoError(t, err) + assert.JSONEq(t, policy, aws.StringValue(got.PolicyText)) + }) + + t.Run("lifecycle policy round-trips", func(t *testing.T) { + policy := `{"rules":[{"rulePriority":1,"selection":{"tagStatus":"untagged","countType":"sinceImagePushed","countUnit":"days","countNumber":14},"action":{"type":"expire"}}]}` + _, err := c.ECR.PutLifecyclePolicy(&ecr.PutLifecyclePolicyInput{ + RepositoryName: aws.String(repo), + LifecyclePolicyText: aws.String(policy), + }) + require.NoError(t, err) + + got, err := c.ECR.GetLifecyclePolicy(&ecr.GetLifecyclePolicyInput{ + RepositoryName: aws.String(repo), + }) + require.NoError(t, err) + assert.JSONEq(t, policy, aws.StringValue(got.LifecyclePolicyText)) + }) + + t.Run("DeleteRepository removes it", func(t *testing.T) { + _, err := c.ECR.DeleteRepository(&ecr.DeleteRepositoryInput{ + RepositoryName: aws.String(repo), + Force: aws.Bool(true), + }) + require.NoError(t, err) + + _, err = c.ECR.DescribeRepositories(&ecr.DescribeRepositoriesInput{ + RepositoryNames: []*string{aws.String(repo)}, + }) + require.Error(t, err, "deleted repo must not be describable") + assert.Contains(t, err.Error(), "RepositoryNotFoundException") + }) +} + +// TestECRGetLoginPassword confirms GetAuthorizationToken returns a usable +// AWS: credential — the input to docker login. +func TestECRGetLoginPassword(t *testing.T) { + f := requireECRFixture(t) + pass := harness.ECRGetLoginPassword(t, f.AWS) + assert.NotEmpty(t, pass) + // JWT shape: three dot-separated segments. + assert.Len(t, splitDots(pass), 3, "authorization password should be a JWT") +} + +func splitDots(s string) []string { + var out []string + start := 0 + for i := 0; i < len(s); i++ { + if s[i] == '.' { + out = append(out, s[start:i]) + start = i + 1 + } + } + return append(out, s[start:]) +} diff --git a/tests/e2e/ecr/main_test.go b/tests/e2e/ecr/main_test.go new file mode 100644 index 000000000..f1bad3cfc --- /dev/null +++ b/tests/e2e/ecr/main_test.go @@ -0,0 +1,71 @@ +//go:build e2e + +// Package ecr is the standalone ECR E2E suite: control-plane CRUD over the +// awsgw ECR surface plus the OCI data plane (docker/crane/skopeo push-pull) +// against a running cluster. The data-plane subtests skip when the client +// binary is missing or the registry host does not resolve, so the suite still +// runs its control-plane leg on a bare runner. +package ecr + +import ( + "os" + "sync" + "testing" + + "github.com/mulgadc/spinifex/tests/e2e/harness" +) + +// Package-scoped singleton fixture, built lazily by requireECRFixture and torn +// down by TestMain after the run. +var ( + pkgFixOnce sync.Once + pkgFix *Fixture + pkgFixErr error +) + +// Fixture carries per-process state shared across every Test* in this package. +type Fixture struct { + Env *harness.Env + AWS *harness.AWSClient + Account string + TmpDir string +} + +func TestMain(m *testing.M) { + code := m.Run() + if pkgFix != nil && pkgFix.TmpDir != "" { + _ = os.RemoveAll(pkgFix.TmpDir) + } + os.Exit(code) +} + +// requireECRFixture returns the package-scoped Fixture, building it on first +// call. Skips the calling test if SPINIFEX_E2E is unset. +func requireECRFixture(t *testing.T) *Fixture { + t.Helper() + pkgFixOnce.Do(func() { + if os.Getenv("SPINIFEX_E2E") == "" { + return + } + env := harness.LoadEnv(t) + awsCli := harness.NewAWSClient(t, env) + tmpDir, terr := os.MkdirTemp("", "ecr-pkgfix-*") + if terr != nil { + pkgFixErr = terr + return + } + pkgFix = &Fixture{ + Env: env, + AWS: awsCli, + Account: harness.IAMAccountID(t, awsCli), + TmpDir: tmpDir, + } + }) + if pkgFixErr != nil { + t.Fatalf("ecr fixture init failed: %v", pkgFixErr) + } + if pkgFix == nil { + t.Skip("ecr fixture unavailable (SPINIFEX_E2E unset)") + } + return pkgFix +} diff --git a/tests/e2e/harness/aws.go b/tests/e2e/harness/aws.go index 0774cc5a4..17a5e9680 100644 --- a/tests/e2e/harness/aws.go +++ b/tests/e2e/harness/aws.go @@ -16,6 +16,7 @@ import ( "github.com/aws/aws-sdk-go/service/acm" "github.com/aws/aws-sdk-go/service/ec2" "github.com/aws/aws-sdk-go/service/ec2/ec2iface" + "github.com/aws/aws-sdk-go/service/ecr" "github.com/aws/aws-sdk-go/service/eks" "github.com/aws/aws-sdk-go/service/elbv2" "github.com/aws/aws-sdk-go/service/iam" @@ -35,6 +36,7 @@ type AWSClient struct { STS *sts.STS EKS *eks.EKS ACM *acm.ACM + ECR *ecr.ECR } // NewAWSClient builds clients pointed at the spinifex gateway using the @@ -131,6 +133,7 @@ func newAWSClient(t *testing.T, env *Env, accessKey, secretKey, sessionToken str STS: sts.New(sess), EKS: eks.New(sess), ACM: acm.New(sess), + ECR: ecr.New(sess), } } @@ -145,5 +148,6 @@ func (c *AWSClient) IgnoreCertErrors() { c.IAM.Config.HTTPClient = hc c.STS.Config.HTTPClient = hc c.ACM.Config.HTTPClient = hc + c.ECR.Config.HTTPClient = hc _ = (*x509.CertPool)(nil) // silence unused-import on hardened paths } diff --git a/tests/e2e/harness/ecr.go b/tests/e2e/harness/ecr.go new file mode 100644 index 000000000..9be92a22a --- /dev/null +++ b/tests/e2e/harness/ecr.go @@ -0,0 +1,112 @@ +//go:build e2e + +package harness + +import ( + "encoding/base64" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ecr" + "github.com/stretchr/testify/require" +) + +// ECRRegistrySuffix is the internal DNS suffix the registry host is built from. +// Defaults to the gateway's DefaultAWSInternalSuffix; override per cluster with +// SPINIFEX_ECR_SUFFIX. +func ECRRegistrySuffix() string { return getenv("SPINIFEX_ECR_SUFFIX", "mulga.internal") } + +// ECRRegistryRegion is the region label in the registry host. Mirrors the AWS +// client region. +func ECRRegistryRegion() string { return getenv("SPINIFEX_AWS_REGION", "ap-southeast-2") } + +// ECRRegistryPort is the awsgw port appended to the registry host for OCI +// clients. Empty by default (production parity, :443); set SPINIFEX_ECR_PORT on +// dev clusters where awsgw binds a non-standard port (e.g. 9999). +func ECRRegistryPort() string { return getenv("SPINIFEX_ECR_PORT", "") } + +// ECRRegistryHost returns {account}.dkr.ecr.{region}.{suffix}[:port] — the OCI +// registry endpoint docker/crane/skopeo address. The bare host must resolve to +// the awsgw bind IP (DNS or /etc/hosts) for the data-plane subtests to run. +func ECRRegistryHost(account string) string { + host := account + ".dkr.ecr." + ECRRegistryRegion() + "." + ECRRegistrySuffix() + if p := ECRRegistryPort(); p != "" { + host += ":" + p + } + return host +} + +// ECRRepositoryURI returns the full push/pull reference for repo under account. +func ECRRepositoryURI(account, repo string) string { + return ECRRegistryHost(account) + "/" + repo +} + +// CreateECRRepository creates repo (idempotent: an existing repo is treated as +// success) and registers best-effort teardown. +func CreateECRRepository(t *testing.T, c *AWSClient, repo string) { + t.Helper() + _, err := c.ECR.CreateRepository(&ecr.CreateRepositoryInput{ + RepositoryName: aws.String(repo), + }) + if err != nil && !strings.Contains(err.Error(), "RepositoryAlreadyExistsException") { + require.NoError(t, err) + } + t.Cleanup(func() { DeleteECRRepositoryBestEffort(c, repo) }) +} + +// DeleteECRRepositoryBestEffort force-deletes repo and all its images, swallowing +// errors — used for cleanup. +func DeleteECRRepositoryBestEffort(c *AWSClient, repo string) { + _, _ = c.ECR.DeleteRepository(&ecr.DeleteRepositoryInput{ + RepositoryName: aws.String(repo), + Force: aws.Bool(true), + }) +} + +// ECRGetLoginPassword returns the decoded registry password (the JWT half of the +// authorization token) for docker login -u AWS --password-stdin. +func ECRGetLoginPassword(t *testing.T, c *AWSClient) string { + t.Helper() + out, err := c.ECR.GetAuthorizationToken(&ecr.GetAuthorizationTokenInput{}) + require.NoError(t, err) + require.NotEmpty(t, out.AuthorizationData, "no authorization data returned") + raw, err := base64.StdEncoding.DecodeString(aws.StringValue(out.AuthorizationData[0].AuthorizationToken)) + require.NoError(t, err) + user, pass, found := strings.Cut(string(raw), ":") + require.True(t, found && user == "AWS", "unexpected authorization token shape") + return pass +} + +// ECRDescribeImageTags returns the set of image tags currently in repo. +func ECRDescribeImageTags(t *testing.T, c *AWSClient, repo string) []string { + t.Helper() + out, err := c.ECR.DescribeImages(&ecr.DescribeImagesInput{ + RepositoryName: aws.String(repo), + }) + require.NoError(t, err) + var tags []string + for _, img := range out.ImageDetails { + for _, tag := range img.ImageTags { + tags = append(tags, aws.StringValue(tag)) + } + } + return tags +} + +// ECRWaitImageTag polls DescribeImages until tag appears in repo or the deadline +// passes. +func ECRWaitImageTag(t *testing.T, c *AWSClient, repo, tag string, timeout time.Duration) bool { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + for _, got := range ECRDescribeImageTags(t, c, repo) { + if got == tag { + return true + } + } + time.Sleep(500 * time.Millisecond) + } + return false +} diff --git a/tests/e2e/harness/registry.go b/tests/e2e/harness/registry.go new file mode 100644 index 000000000..0d1acfd1c --- /dev/null +++ b/tests/e2e/harness/registry.go @@ -0,0 +1,115 @@ +//go:build e2e + +package harness + +import ( + "bytes" + "net" + "os" + "os/exec" + "strings" + "testing" + "time" +) + +// ExternalCLI wraps an OCI client binary (docker/crane/skopeo) for the ECR +// data-plane subtests. Env is appended to the process environment (e.g. +// DOCKER_CONFIG to isolate the auth file). +type ExternalCLI struct { + Name string + Bin string + Env []string +} + +// newExternalCLI locates bin on PATH (overridable via _BIN, upper-cased) +// and SKIPS the calling test when absent — the data-plane subtests are optional +// on runners without the client installed. +func newExternalCLI(t *testing.T, name string, env []string) *ExternalCLI { + t.Helper() + bin := os.Getenv(strings.ToUpper(name) + "_BIN") + if bin == "" { + bin = name + } + resolved, err := exec.LookPath(bin) + if err != nil { + t.Skipf("%s not found on PATH (%v); skipping data-plane subtest", name, err) + } + return &ExternalCLI{Name: name, Bin: resolved, Env: env} +} + +// NewDocker / NewCrane / NewSkopeo build wrappers for each client, skipping the +// test if the binary is missing. dockerConfig isolates the docker auth file so +// crane/skopeo can reuse it via DOCKER_CONFIG without touching ~/.docker. +func NewDocker(t *testing.T, dockerConfig string) *ExternalCLI { + return newExternalCLI(t, "docker", []string{"DOCKER_CONFIG=" + dockerConfig}) +} + +func NewCrane(t *testing.T, dockerConfig string) *ExternalCLI { + return newExternalCLI(t, "crane", []string{"DOCKER_CONFIG=" + dockerConfig}) +} + +func NewSkopeo(t *testing.T, dockerConfig string) *ExternalCLI { + return newExternalCLI(t, "skopeo", []string{"REGISTRY_AUTH_FILE=" + dockerConfig + "/config.json"}) +} + +// Run executes the client with args and returns combined output. Errors are +// returned (not fatal) so callers can assert on output or poll. +func (c *ExternalCLI) Run(timeout time.Duration, args ...string) (string, error) { + cmd := exec.Command(c.Bin, args...) //nolint:gosec // bin is LookPath-resolved, args test-controlled + cmd.Env = append(os.Environ(), c.Env...) + var buf bytes.Buffer + cmd.Stdout = &buf + cmd.Stderr = &buf + if err := cmd.Start(); err != nil { + return "", err + } + done := make(chan error, 1) + go func() { done <- cmd.Wait() }() + select { + case err := <-done: + return buf.String(), err + case <-time.After(timeout): + _ = cmd.Process.Kill() + <-done + return buf.String(), &timeoutError{} + } +} + +// RunStdin is Run with data piped to the process stdin (docker login +// --password-stdin). +func (c *ExternalCLI) RunStdin(timeout time.Duration, stdin string, args ...string) (string, error) { + cmd := exec.Command(c.Bin, args...) //nolint:gosec // bin is LookPath-resolved, args test-controlled + cmd.Env = append(os.Environ(), c.Env...) + cmd.Stdin = strings.NewReader(stdin) + var buf bytes.Buffer + cmd.Stdout = &buf + cmd.Stderr = &buf + if err := cmd.Start(); err != nil { + return "", err + } + done := make(chan error, 1) + go func() { done <- cmd.Wait() }() + select { + case err := <-done: + return buf.String(), err + case <-time.After(timeout): + _ = cmd.Process.Kill() + <-done + return buf.String(), &timeoutError{} + } +} + +// RequireRegistryResolves SKIPS the calling test unless host resolves to an +// address. host may carry a :port suffix (stripped before lookup). The +// data-plane subtests need {account}.dkr.ecr.{region}.{suffix} to reach the +// awsgw bind IP (Route53 in prod, /etc/hosts in dev); without it the client +// cannot connect. +func RequireRegistryResolves(t *testing.T, host string) { + t.Helper() + if h, _, err := net.SplitHostPort(host); err == nil { + host = h + } + if _, err := net.LookupHost(host); err != nil { + t.Skipf("registry host %q does not resolve (%v); seed /etc/hosts or DNS", host, err) + } +}