diff --git a/spinifex/domains/ec2/awsapi/instance/DescribeInstanceTypes.go b/spinifex/domains/ec2/awsapi/instance/DescribeInstanceTypes.go index 9256b0614..ad1ef5b2f 100644 --- a/spinifex/domains/ec2/awsapi/instance/DescribeInstanceTypes.go +++ b/spinifex/domains/ec2/awsapi/instance/DescribeInstanceTypes.go @@ -12,6 +12,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" + instancesvc "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance" "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" filterutil "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters" "github.com/nats-io/nats.go" @@ -43,18 +44,24 @@ func DescribeInstanceTypes(ctx context.Context, input *ec2.DescribeInstanceTypes } var allInstanceTypes []*ec2.InstanceTypeInfo + availability := make(map[string]int64) answered := make(map[string]bool, len(frames)) + usable := 0 for _, frame := range frames { - var nodeOutput ec2.DescribeInstanceTypesOutput + var nodeOutput instancesvc.DescribeInstanceTypesNodeOutput if json.Unmarshal(frame.Data, &nodeOutput) == nil { allInstanceTypes = append(allInstanceTypes, nodeOutput.InstanceTypes...) + for name, count := range nodeOutput.Availability { + availability[name] += count + } + usable++ if frame.NodeID != "" { answered[frame.NodeID] = true } } } - // capacity=true filter shows all slots (including duplicates) across nodes. + // capacity=true reports cluster-wide free slots per type. showCapacity := false for _, f := range input.Filters { if f.Name != nil && *f.Name == "capacity" { @@ -76,7 +83,17 @@ func DescribeInstanceTypes(ctx context.Context, input *ec2.DescribeInstanceTypes var finalInstanceTypes []*ec2.InstanceTypeInfo if showCapacity { - finalInstanceTypes = allInstanceTypes + // No node answered, so nothing is known about capacity. Reporting an + // empty list here reads as "no capacity anywhere", which is what made + // the console refuse every launch. + if usable == 0 { + return nil, awserrors.WithRetryAfter( + awserrors.Errorf(awserrors.ErrorServiceUnavailable, + "cannot determine cluster capacity: no node answered the describe fan-out"), + describeIncompleteRetryAfter, + ) + } + finalInstanceTypes = expandAvailability(allInstanceTypes, availability) } else { seen := make(map[string]bool) for _, it := range allInstanceTypes { @@ -143,6 +160,34 @@ func DescribeInstanceTypes(ctx context.Context, input *ec2.DescribeInstanceTypes return output, nil } +// expandAvailability returns one entry per free cluster-wide slot for each +// type, deduplicating the per-node descriptions first. +// +// The client SDK drops response fields it does not model, so a count cannot +// reach the console any other way. Nodes send counts, and only this last hop +// repeats the entry. +func expandAvailability(types []*ec2.InstanceTypeInfo, availability map[string]int64) []*ec2.InstanceTypeInfo { + byName := make(map[string]*ec2.InstanceTypeInfo, len(types)) + order := make([]string, 0, len(types)) + for _, it := range types { + if it == nil || it.InstanceType == nil { + continue + } + if _, seen := byName[*it.InstanceType]; !seen { + byName[*it.InstanceType] = it + order = append(order, *it.InstanceType) + } + } + + expanded := make([]*ec2.InstanceTypeInfo, 0, len(order)) + for _, name := range order { + for range availability[name] { + expanded = append(expanded, byName[name]) + } + } + return expanded +} + // missingInstanceTypes returns the requested types absent from found, in the // caller's order and without repeats. func missingInstanceTypes(found []*ec2.InstanceTypeInfo, requested []*string) []string { diff --git a/spinifex/domains/ec2/awsapi/instance/DescribeInstanceTypes_test.go b/spinifex/domains/ec2/awsapi/instance/DescribeInstanceTypes_test.go index db696c840..dc899f674 100644 --- a/spinifex/domains/ec2/awsapi/instance/DescribeInstanceTypes_test.go +++ b/spinifex/domains/ec2/awsapi/instance/DescribeInstanceTypes_test.go @@ -7,6 +7,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" + instancesvc "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance" "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" @@ -87,54 +88,81 @@ func TestDescribeInstanceTypes_DeduplicatesAcrossNodes(t *testing.T) { assert.Equal(t, 1, seen["m5.large"]) } -func TestDescribeInstanceTypes_CapacityFilterShowsDuplicates(t *testing.T) { +func TestDescribeInstanceTypes_CapacityFilterSumsPerNodeCounts(t *testing.T) { t.Parallel() _, nc := startTestNATSServer(t) // Node 1 nc.Subscribe("ec2.DescribeInstanceTypes", func(msg *nats.Msg) { - data, _ := json.Marshal(&ec2.DescribeInstanceTypesOutput{ - InstanceTypes: []*ec2.InstanceTypeInfo{ - {InstanceType: aws.String("t3.micro")}, - }, - }) - msg.Respond(data) + msg.Respond(capacityPayload(t, map[string]int64{"t3.micro": 1})) }) - // Node 2 reports same type + // Node 2 reports the same type nc2, err := nats.Connect(nc.ConnectedUrl()) require.NoError(t, err) defer nc2.Close() nc2.Subscribe("ec2.DescribeInstanceTypes", func(msg *nats.Msg) { - data, _ := json.Marshal(&ec2.DescribeInstanceTypesOutput{ - InstanceTypes: []*ec2.InstanceTypeInfo{ - {InstanceType: aws.String("t3.micro")}, - }, - }) - msg.Respond(data) + msg.Respond(capacityPayload(t, map[string]int64{"t3.micro": 1})) }) nc.Flush() nc2.Flush() - // With capacity filter = true, duplicates should be preserved - input := &ec2.DescribeInstanceTypesInput{ - Filters: []*ec2.Filter{ - { - Name: aws.String("capacity"), - Values: []*string{aws.String("true")}, - }, - }, - } - output, err := DescribeInstanceTypes(context.Background(), input, nc, 2, nil, "") + output, err := DescribeInstanceTypes(context.Background(), capacityInput(), nc, 2, nil, "") require.NoError(t, err) require.NotNil(t, output) - // Both slots should appear since capacity filter is enabled + // One entry per free slot across both nodes. assert.Len(t, output.InstanceTypes, 2) } +// A node reports a count, not one entry per slot, so the reply stays the same +// size whether the host is idle or full. +func TestDescribeInstanceTypes_CapacityIsBoundedByTypeCountNotFreeSlots(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + + payload := capacityPayload(t, map[string]int64{"t3.micro": 4096}) + require.Less(t, len(payload), 4096, "the node reply must not grow with free capacity") + + subscribeAsNode(t, nc, "ec2.DescribeInstanceTypes", "n1", payload) + + output, err := DescribeInstanceTypes(context.Background(), capacityInput(), nc, 1, nil, "") + require.NoError(t, err) + assert.Len(t, output.InstanceTypes, 4096) +} + +// Nothing is known about capacity when no node answered, so the request fails +// rather than reporting an empty list the console reads as no capacity. +func TestDescribeInstanceTypes_CapacityWithNoResponsesIsRetryable(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + + _, err := DescribeInstanceTypes(context.Background(), capacityInput(), nc, 0, nil, "") + + code, _, ok := awserrors.ResolveErrorDetail(err) + require.True(t, ok) + assert.Equal(t, awserrors.ErrorServiceUnavailable, code) +} + +func capacityInput() *ec2.DescribeInstanceTypesInput { + return &ec2.DescribeInstanceTypesInput{ + Filters: []*ec2.Filter{{Name: aws.String("capacity"), Values: []*string{aws.String("true")}}}, + } +} + +func capacityPayload(t *testing.T, availability map[string]int64) []byte { + t.Helper() + out := &instancesvc.DescribeInstanceTypesNodeOutput{Availability: availability} + for name := range availability { + out.InstanceTypes = append(out.InstanceTypes, &ec2.InstanceTypeInfo{InstanceType: aws.String(name)}) + } + data, err := json.Marshal(out) + require.NoError(t, err) + return data +} + func TestDescribeInstanceTypes_CapacityFilterFalseDeduplicates(t *testing.T) { t.Parallel() _, nc := startTestNATSServer(t) diff --git a/spinifex/domains/ec2/instance/service.go b/spinifex/domains/ec2/instance/service.go index 87d901939..f5342f527 100644 --- a/spinifex/domains/ec2/instance/service.go +++ b/spinifex/domains/ec2/instance/service.go @@ -17,7 +17,7 @@ type InstanceService interface { RunInstances(ctx context.Context, input *ec2.RunInstancesInput, accountID string) (*ec2.Reservation, error) DescribeInstances(ctx context.Context, input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) DescribeInstanceStatus(ctx context.Context, input *ec2.DescribeInstanceStatusInput, accountID string) (*ec2.DescribeInstanceStatusOutput, error) - DescribeInstanceTypes(ctx context.Context, input *ec2.DescribeInstanceTypesInput, accountID string) (*ec2.DescribeInstanceTypesOutput, error) + DescribeInstanceTypes(ctx context.Context, input *ec2.DescribeInstanceTypesInput, accountID string) (*DescribeInstanceTypesNodeOutput, error) DescribeInstanceAttribute(ctx context.Context, input *ec2.DescribeInstanceAttributeInput, accountID string) (*ec2.DescribeInstanceAttributeOutput, error) DescribeStoppedInstances(ctx context.Context, input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) DescribeTerminatedInstances(ctx context.Context, input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) @@ -49,11 +49,30 @@ type TerminateStoppedInstanceOutput struct { InstanceID string `json:"instanceId"` } +// DescribeInstanceTypesNodeOutput is one node's reply to the describe fan-out. +// Availability is a count per instance type, so the reply is bounded by the +// number of types and never by how much capacity is free. +type DescribeInstanceTypesNodeOutput struct { + InstanceTypes []*ec2.InstanceTypeInfo `json:"InstanceTypes"` + + // Set only for the capacity request, keyed by instance type name. A type + // present in InstanceTypes with no entry here has no free slot on the node. + Availability map[string]int64 `json:"Availability,omitempty"` +} + +// InstanceTypeAvailability pairs an instance type with the number of instances +// of it the node can currently admit. +type InstanceTypeAvailability struct { + Info *ec2.InstanceTypeInfo + Available int64 +} + // ResourceCapacityProvider exposes per-node instance-type availability. -// GetAvailableInstanceTypeInfos gates on free capacity; GetSupportedInstanceTypeInfos -// returns all configured types regardless of allocation. +// GetInstanceTypeAvailability counts the free slots per type; +// GetSupportedInstanceTypeInfos returns all configured types regardless of +// allocation. type ResourceCapacityProvider interface { - GetAvailableInstanceTypeInfos(showCapacity bool) []*ec2.InstanceTypeInfo + GetInstanceTypeAvailability() []InstanceTypeAvailability GetSupportedInstanceTypeInfos() []*ec2.InstanceTypeInfo } diff --git a/spinifex/domains/ec2/instance/service_impl.go b/spinifex/domains/ec2/instance/service_impl.go index e591d67bc..055f56bb1 100644 --- a/spinifex/domains/ec2/instance/service_impl.go +++ b/spinifex/domains/ec2/instance/service_impl.go @@ -2141,9 +2141,9 @@ func (s *InstanceServiceImpl) DescribeInstances(ctx context.Context, input *ec2. } // DescribeInstanceTypes returns supported instance types. With filter -// capacity=true, expands each type to one entry per free slot for cluster-wide -// capacity aggregation. -func (s *InstanceServiceImpl) DescribeInstanceTypes(ctx context.Context, input *ec2.DescribeInstanceTypesInput, _ string) (*ec2.DescribeInstanceTypesOutput, error) { +// capacity=true, each admissible type is returned once with the number of +// instances of it this node can take, for cluster-wide capacity aggregation. +func (s *InstanceServiceImpl) DescribeInstanceTypes(ctx context.Context, input *ec2.DescribeInstanceTypesInput, _ string) (*DescribeInstanceTypesNodeOutput, error) { slog.InfoContext(ctx, "Processing DescribeInstanceTypes request from this node") if s.resourceMgr == nil { @@ -2162,14 +2162,30 @@ func (s *InstanceServiceImpl) DescribeInstanceTypes(ctx context.Context, input * } } - var filteredTypes []*ec2.InstanceTypeInfo - if showCapacity { - filteredTypes = s.resourceMgr.GetAvailableInstanceTypeInfos(true) - } else { - filteredTypes = s.resourceMgr.GetSupportedInstanceTypeInfos() + if !showCapacity { + types := s.resourceMgr.GetSupportedInstanceTypeInfos() + slog.InfoContext(ctx, "DescribeInstanceTypes completed", "count", len(types), "showCapacity", false) + return &DescribeInstanceTypesNodeOutput{InstanceTypes: types}, nil } - slog.InfoContext(ctx, "DescribeInstanceTypes completed", "count", len(filteredTypes), "showCapacity", showCapacity) - return &ec2.DescribeInstanceTypesOutput{InstanceTypes: filteredTypes}, nil + + avail := s.resourceMgr.GetInstanceTypeAvailability() + out := &DescribeInstanceTypesNodeOutput{ + InstanceTypes: make([]*ec2.InstanceTypeInfo, 0, len(avail)), + Availability: make(map[string]int64, len(avail)), + } + totalSlots := int64(0) + for _, a := range avail { + if a.Info == nil || a.Info.InstanceType == nil || a.Available <= 0 { + continue + } + out.InstanceTypes = append(out.InstanceTypes, a.Info) + out.Availability[*a.Info.InstanceType] = a.Available + totalSlots += a.Available + } + + slog.InfoContext(ctx, "DescribeInstanceTypes completed", "count", len(out.InstanceTypes), + "total_available_slots", totalSlots, "showCapacity", true) + return out, nil } // DescribeStoppedInstances returns stopped instances from shared KV. diff --git a/spinifex/domains/ec2/instance/service_impl_test.go b/spinifex/domains/ec2/instance/service_impl_test.go index dc7a0eb76..43426e9cc 100644 --- a/spinifex/domains/ec2/instance/service_impl_test.go +++ b/spinifex/domains/ec2/instance/service_impl_test.go @@ -693,6 +693,7 @@ type fakeResourceCapacityProvider struct { allocated []*ec2.InstanceTypeInfo deallocated []*ec2.InstanceTypeInfo canAllocFn func(*ec2.InstanceTypeInfo, int) int + availableFn func(*ec2.InstanceTypeInfo) int64 reservationAllocErr error reservationAllocated []*ec2.InstanceTypeInfo @@ -704,10 +705,18 @@ type fakeResourceCapacityProvider struct { func (f *fakeResourceCapacityProvider) HostUnderMemoryPressure() bool { return f.memPressure } -func (f *fakeResourceCapacityProvider) GetAvailableInstanceTypeInfos(showCapacity bool) []*ec2.InstanceTypeInfo { +func (f *fakeResourceCapacityProvider) GetInstanceTypeAvailability() []InstanceTypeAvailability { f.calls++ - f.gotShowCap = showCapacity - return f.types + f.gotShowCap = true + avail := make([]InstanceTypeAvailability, 0, len(f.types)) + for _, it := range f.types { + slots := int64(1) + if f.availableFn != nil { + slots = f.availableFn(it) + } + avail = append(avail, InstanceTypeAvailability{Info: it, Available: slots}) + } + return avail } func (f *fakeResourceCapacityProvider) GetSupportedInstanceTypeInfos() []*ec2.InstanceTypeInfo { diff --git a/spinifex/domains/ec2/instance/service_nats.go b/spinifex/domains/ec2/instance/service_nats.go index b90964637..b05fb2236 100644 --- a/spinifex/domains/ec2/instance/service_nats.go +++ b/spinifex/domains/ec2/instance/service_nats.go @@ -41,7 +41,7 @@ func (s *NATSInstanceService) DescribeInstances(_ context.Context, _ *ec2.Descri // DescribeInstanceTypes is a fan-out operation; use the gateway's scatter-gather // helper instead of this single-node client. -func (s *NATSInstanceService) DescribeInstanceTypes(_ context.Context, _ *ec2.DescribeInstanceTypesInput, _ string) (*ec2.DescribeInstanceTypesOutput, error) { +func (s *NATSInstanceService) DescribeInstanceTypes(_ context.Context, _ *ec2.DescribeInstanceTypesInput, _ string) (*DescribeInstanceTypesNodeOutput, error) { return nil, fmt.Errorf("DescribeInstanceTypes is fan-out; use gateway scatter-gather, not NATSInstanceService") } diff --git a/spinifex/runtime/daemon/daemon.go b/spinifex/runtime/daemon/daemon.go index d4c1f61a0..60ce520b8 100644 --- a/spinifex/runtime/daemon/daemon.go +++ b/spinifex/runtime/daemon/daemon.go @@ -604,14 +604,19 @@ func (rm *ResourceManager) GetInstanceTypeInfos() []*ec2.InstanceTypeInfo { return infos } -// GetAvailableInstanceTypeInfos returns instance types based on total host capacity. -// If showCapacity is true, it returns multiple entries representing available slots. -// If showCapacity is false, it returns each supported type only once. -func (rm *ResourceManager) GetAvailableInstanceTypeInfos(showCapacity bool) []*ec2.InstanceTypeInfo { +// GetInstanceTypeAvailability returns each admissible instance type once with +// the number of instances of it this node can currently take. Types with no +// free slot are omitted. +// +// The count is returned as a number rather than as one entry per free slot. +// Repeating the entry made the reply grow with free capacity, so an idle host +// could exceed the NATS maximum payload and report no capacity at all. +func (rm *ResourceManager) GetInstanceTypeAvailability() []ec2instance.InstanceTypeAvailability { rm.mu.RLock() defer rm.mu.RUnlock() - var infos []*ec2.InstanceTypeInfo + avail := make([]ec2instance.InstanceTypeAvailability, 0, len(rm.instanceTypes)) + totalSlots := int64(0) for name, it := range rm.instanceTypes { if instancetypes.IsSystemType(name) { @@ -625,44 +630,37 @@ func (rm *ResourceManager) GetAvailableInstanceTypeInfos(showCapacity bool) []*e continue } + var count int64 + // GPU types are capacity-gated by GPU pool size, not host CPU/memory. // The GPU is the scarce resource; CPU/memory on GPU-class hardware is abundant. if instancetypes.IsGPUType(it) { - count := rm.admissibleGPUInstances(name) - if showCapacity { - for range count { - infos = append(infos, it) - } - } else if count > 0 { - infos = append(infos, it) - } + count = int64(rm.admissibleGPUInstances(name)) + } else { + count = int64(canAllocateCount( + rm.hostVCPU-rm.reservedVCPU-rm.reservedCRVCPU, rm.allocatedVCPU, + rm.hostMemGB-rm.reservedMem-rm.reservedCRMem, rm.allocatedMem, + vCPUs, rm.instanceMemChargeMiB(it), + 1<<30, // effectively unlimited — let resources be the constraint + 0, false, + )) + } + + if count <= 0 { continue } - count := canAllocateCount( - rm.hostVCPU-rm.reservedVCPU-rm.reservedCRVCPU, rm.allocatedVCPU, - rm.hostMemGB-rm.reservedMem-rm.reservedCRMem, rm.allocatedMem, - vCPUs, rm.instanceMemChargeMiB(it), - 1<<30, // effectively unlimited — let resources be the constraint - 0, false, - ) - - if showCapacity { - for range count { - infos = append(infos, it) - } - } else if count > 0 { - infos = append(infos, it) - } + avail = append(avail, ec2instance.InstanceTypeAvailability{Info: it, Available: count}) + totalSlots += count } - slog.Info("GetAvailableInstanceTypeInfos", "total_types", len(rm.instanceTypes), "total_available_slots", len(infos), + slog.Info("GetInstanceTypeAvailability", "total_types", len(rm.instanceTypes), + "available_types", len(avail), "total_available_slots", totalSlots, "hostVCPU", rm.hostVCPU, "hostMem", rm.hostMemGB, "reservedVCPU", rm.reservedVCPU, "reservedMem", rm.reservedMem, - "reservedCRVCPU", rm.reservedCRVCPU, "reservedCRMem", rm.reservedCRMem, - "showCapacity", showCapacity) + "reservedCRVCPU", rm.reservedCRVCPU, "reservedCRMem", rm.reservedCRMem) - return infos + return avail } // GetSupportedInstanceTypeInfos returns every supported instance type regardless diff --git a/spinifex/runtime/daemon/daemon_test.go b/spinifex/runtime/daemon/daemon_test.go index bee7ea76b..1bb25d5a3 100644 --- a/spinifex/runtime/daemon/daemon_test.go +++ b/spinifex/runtime/daemon/daemon_test.go @@ -297,14 +297,14 @@ func TestGetInstanceTypeInfos(t *testing.T) { } } -// TestGetAvailableInstanceTypeInfos_ResourceFiltering tests that instance types are filtered by available resources. -func TestGetAvailableInstanceTypeInfos_ResourceFiltering(t *testing.T) { +// TestGetInstanceTypeAvailability_ResourceFiltering tests that instance types are filtered by available resources. +func TestGetInstanceTypeAvailability_ResourceFiltering(t *testing.T) { rm, err := NewResourceManager(nil, nil, nil) require.NoError(t, err) // Get initial count of all available types allTypes := rm.GetInstanceTypeInfos() - initialAvailable := rm.GetAvailableInstanceTypeInfos(false) + initialAvailable := availableTypeInfos(rm, false) t.Logf("System has %d vCPUs, %.2f GB RAM (reserved: %d vCPU, %.2f GB)", rm.hostVCPU, rm.hostMemGB, rm.reservedVCPU, rm.reservedMem) @@ -351,7 +351,7 @@ func TestGetAvailableInstanceTypeInfos_ResourceFiltering(t *testing.T) { nanoKey, rm.allocatedVCPU, rm.allocatedMem) // Now get available types - should be fewer or equal (depending on system resources) - afterAllocation := rm.GetAvailableInstanceTypeInfos(false) + afterAllocation := availableTypeInfos(rm, false) t.Logf("Available after allocation: %d", len(afterAllocation)) // Verify all returned types fit within REMAINING schedulable resources @@ -374,7 +374,7 @@ func TestGetAvailableInstanceTypeInfos_ResourceFiltering(t *testing.T) { // Deallocate and verify we get the same available types as before rm.deallocate(nanoType) - afterDeallocation := rm.GetAvailableInstanceTypeInfos(false) + afterDeallocation := availableTypeInfos(rm, false) assert.Len(t, afterDeallocation, len(initialAvailable), "Should have same available types after deallocation") } @@ -562,7 +562,7 @@ func TestHandleEC2DescribeInstanceTypes(t *testing.T) { // type near the schedulable-memory boundary that the allocator then // rejects, flaking the test on small hosts. var instanceType2CPU *ec2.InstanceTypeInfo - for _, it := range daemon.resourceMgr.GetAvailableInstanceTypeInfos(true) { + for _, it := range availableTypeInfos(daemon.resourceMgr, true) { if it.VCpuInfo == nil || it.VCpuInfo.DefaultVCpus == nil || *it.VCpuInfo.DefaultVCpus != 2 || it.MemoryInfo == nil || it.MemoryInfo.SizeInMiB == nil { continue @@ -645,7 +645,7 @@ func TestHandleEC2DescribeInstanceTypes(t *testing.T) { reply, err := daemon.natsConn.Request("ec2.DescribeInstanceTypes", msgData, 5*time.Second) require.NoError(t, err) - var output ec2.DescribeInstanceTypesOutput + var output ec2instance.DescribeInstanceTypesNodeOutput err = json.Unmarshal(reply.Data, &output) require.NoError(t, err) @@ -676,13 +676,7 @@ func TestHandleEC2DescribeInstanceTypes(t *testing.T) { err = json.Unmarshal(reply.Data, &output) require.NoError(t, err) - // Verify duplicate slots exist — find a nano type and confirm it has 2 slots - typeCounts := make(map[string]int) - for _, info := range output.InstanceTypes { - if info.InstanceType != nil { - typeCounts[*info.InstanceType]++ - } - } + // Verify the count rises with capacity — find a nano type and confirm 2 slots // Find any nano type in the generated types var nanoType string for name := range daemon.resourceMgr.instanceTypes { @@ -692,7 +686,7 @@ func TestHandleEC2DescribeInstanceTypes(t *testing.T) { } } require.NotEmpty(t, nanoType, "Should have at least one nano type") - assert.Equal(t, 2, typeCounts[nanoType], "Should have 2 slots for %s with 4 vCPUs", nanoType) + assert.Equal(t, int64(2), output.Availability[nanoType], "Should have 2 slots for %s with 4 vCPUs", nanoType) }) } @@ -2995,9 +2989,25 @@ func TestDaemon_LoadState_CorruptFileFatal(t *testing.T) { assert.Contains(t, err.Error(), "read local state") } -// --- GetAvailableInstanceTypeInfos edge cases --- +// availableTypeInfos expands the availability counts the way the gateway does, +// so these tests can keep asserting on free slots or on distinct types. +func availableTypeInfos(rm *ResourceManager, showCapacity bool) []*ec2.InstanceTypeInfo { + var infos []*ec2.InstanceTypeInfo + for _, a := range rm.GetInstanceTypeAvailability() { + if !showCapacity { + infos = append(infos, a.Info) + continue + } + for range a.Available { + infos = append(infos, a.Info) + } + } + return infos +} -func TestGetAvailableInstanceTypeInfos_Overcommitted(t *testing.T) { +// --- GetInstanceTypeAvailability edge cases --- + +func TestGetInstanceTypeAvailability_Overcommitted(t *testing.T) { rm := &ResourceManager{ hostVCPU: 4, hostMemGB: 8.0, @@ -3012,14 +3022,14 @@ func TestGetAvailableInstanceTypeInfos_Overcommitted(t *testing.T) { }, } - infos := rm.GetAvailableInstanceTypeInfos(true) + infos := availableTypeInfos(rm, true) assert.Empty(t, infos, "overcommitted resources should return 0 available slots") - infos = rm.GetAvailableInstanceTypeInfos(false) + infos = availableTypeInfos(rm, false) assert.Empty(t, infos) } -func TestGetAvailableInstanceTypeInfos_ShowCapacity(t *testing.T) { +func TestGetInstanceTypeAvailability_ShowCapacity(t *testing.T) { rm := &ResourceManager{ hostVCPU: 8, hostMemGB: 16.0, @@ -3035,15 +3045,15 @@ func TestGetAvailableInstanceTypeInfos_ShowCapacity(t *testing.T) { } // With showCapacity=true, should return multiple entries - infos := rm.GetAvailableInstanceTypeInfos(true) + infos := availableTypeInfos(rm, true) assert.Greater(t, len(infos), 1) // With showCapacity=false, should return exactly 1 - infos = rm.GetAvailableInstanceTypeInfos(false) + infos = availableTypeInfos(rm, false) assert.Len(t, infos, 1) } -// --- GPU-gated GetAvailableInstanceTypeInfos --- +// --- GPU-gated GetInstanceTypeAvailability --- func makeGPUInstanceType(name string, vcpus int64, memMiB int64) *ec2.InstanceTypeInfo { return &ec2.InstanceTypeInfo{ @@ -3066,7 +3076,7 @@ func makeGPUDevice() gpu.GPUDevice { } // GPU types with no gpuManager are hidden — there is no GPU hardware on this node. -func TestGetAvailableInstanceTypeInfos_GPUType_NoManager_Hidden(t *testing.T) { +func TestGetInstanceTypeAvailability_GPUType_NoManager_Hidden(t *testing.T) { rm := &ResourceManager{ hostVCPU: 128, hostMemGB: 1024.0, @@ -3074,13 +3084,13 @@ func TestGetAvailableInstanceTypeInfos_GPUType_NoManager_Hidden(t *testing.T) { "g7e.4xlarge": makeGPUInstanceType("g7e.4xlarge", 16, 128*1024), }, } - infos := rm.GetAvailableInstanceTypeInfos(false) + infos := availableTypeInfos(rm, false) assert.Empty(t, infos, "GPU type must not appear when gpuManager is nil") } // GPU types gate on pool availability, not host CPU/memory — even a tiny host // with a GPU pool exposes the type. -func TestGetAvailableInstanceTypeInfos_GPUType_GatedByPool(t *testing.T) { +func TestGetInstanceTypeAvailability_GPUType_GatedByPool(t *testing.T) { mgr := gpu.NewManager([]gpu.GPUDevice{makeGPUDevice(), makeGPUDevice()}) rm := &ResourceManager{ hostVCPU: 4, // would normally block a 16-vCPU instance @@ -3090,12 +3100,12 @@ func TestGetAvailableInstanceTypeInfos_GPUType_GatedByPool(t *testing.T) { }, gpuManager: mgr, } - infos := rm.GetAvailableInstanceTypeInfos(false) + infos := availableTypeInfos(rm, false) assert.Len(t, infos, 1, "GPU type must appear when pool has free GPUs regardless of CPU/memory") } // Multi-GPU type (g7e.12xlarge, needs 2) appears only when pool has ≥ 2 free GPUs. -func TestGetAvailableInstanceTypeInfos_MultiGPUType_RequiresTwoGPUs(t *testing.T) { +func TestGetInstanceTypeAvailability_MultiGPUType_RequiresTwoGPUs(t *testing.T) { oneGPU := gpu.NewManager([]gpu.GPUDevice{makeGPUDevice()}) rm := &ResourceManager{ hostVCPU: 128, @@ -3105,17 +3115,17 @@ func TestGetAvailableInstanceTypeInfos_MultiGPUType_RequiresTwoGPUs(t *testing.T }, gpuManager: oneGPU, } - infos := rm.GetAvailableInstanceTypeInfos(false) + infos := availableTypeInfos(rm, false) assert.Empty(t, infos, "g7e.12xlarge must not appear with only 1 free GPU") twoGPUs := gpu.NewManager([]gpu.GPUDevice{makeGPUDevice(), makeGPUDevice()}) rm.gpuManager = twoGPUs - infos = rm.GetAvailableInstanceTypeInfos(false) + infos = availableTypeInfos(rm, false) assert.Len(t, infos, 1, "g7e.12xlarge must appear when pool has ≥ 2 free GPUs") } // showCapacity=true for GPU types emits pool/gpuCount slots. -func TestGetAvailableInstanceTypeInfos_GPUType_ShowCapacity(t *testing.T) { +func TestGetInstanceTypeAvailability_GPUType_ShowCapacity(t *testing.T) { mgr := gpu.NewManager([]gpu.GPUDevice{makeGPUDevice(), makeGPUDevice(), makeGPUDevice(), makeGPUDevice()}) rm := &ResourceManager{ hostVCPU: 128, @@ -3127,7 +3137,7 @@ func TestGetAvailableInstanceTypeInfos_GPUType_ShowCapacity(t *testing.T) { gpuManager: mgr, } // 4 GPUs → 4× g7e.4xlarge (1 GPU each) or 2× g7e.12xlarge (2 GPUs each) - infos := rm.GetAvailableInstanceTypeInfos(true) + infos := availableTypeInfos(rm, true) count4xl := 0 count12xl := 0 for _, it := range infos { @@ -3212,7 +3222,7 @@ func TestGetSupportedInstanceTypeInfos_IgnoresFreeSlots(t *testing.T) { } // Capacity-gated view: nothing fits because resources are exhausted. - assert.Empty(t, rm.GetAvailableInstanceTypeInfos(false), + assert.Empty(t, availableTypeInfos(rm, false), "capacity-gated set must be empty when resources are exhausted") // Supported view: every loaded type is still announced. diff --git a/spinifex/runtime/daemon/describe_instance_types_payload_test.go b/spinifex/runtime/daemon/describe_instance_types_payload_test.go new file mode 100644 index 000000000..fd0a295a1 --- /dev/null +++ b/spinifex/runtime/daemon/describe_instance_types_payload_test.go @@ -0,0 +1,56 @@ +package daemon + +import ( + "context" + "encoding/json" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + ec2instance "github.com/mulgadc/spinifex/spinifex/domains/ec2/instance" + "github.com/mulgadc/spinifex/spinifex/domains/ec2/instancetypes" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// natsMaxPayload is the server's configured maximum message size. A reply +// larger than this is dropped, and the fan-out then reports no capacity. +const natsMaxPayload = 1 << 20 + +// A node's capacity reply is bounded by how many instance types exist, not by +// how much capacity is free, so an idle host answers as compactly as a full +// one. An earlier form sent one copy of the type per free slot, which an idle +// host grew past the NATS maximum payload. +func TestDescribeInstanceTypesCapacityReplyFitsTheNATSPayloadWhenHostIsIdle(t *testing.T) { + catalogue := instancetypes.DetectAndGenerate(instancetypes.HostCPU{}, "x86_64", nil) + require.NotEmpty(t, catalogue) + + // A host far larger than any in this fleet, entirely unallocated, is the + // worst case for a reply whose size tracks free capacity. + rm := &ResourceManager{ + hostVCPU: 4096, + hostMemGB: 16384, + instanceTypes: catalogue, + } + + svc := ec2instance.NewInstanceServiceImpl(nil, catalogue, nil, nil, nil, rm, nil) + + out, err := svc.DescribeInstanceTypes(context.Background(), &ec2.DescribeInstanceTypesInput{ + Filters: []*ec2.Filter{{Name: aws.String("capacity"), Values: []*string{aws.String("true")}}}, + }, "") + require.NoError(t, err) + + var totalSlots int64 + for _, n := range out.Availability { + totalSlots += n + } + require.Greater(t, totalSlots, int64(natsMaxPayload/64), + "the host must be idle enough that a per-slot reply would overflow") + + data, err := json.Marshal(out) + require.NoError(t, err) + assert.Less(t, len(data), natsMaxPayload, + "capacity reply is %d bytes for %d free slots across %d types", + len(data), totalSlots, len(out.InstanceTypes)) + assert.Len(t, out.Availability, len(out.InstanceTypes)) +} diff --git a/spinifex/runtime/roles/spinifexui/frontend/e2e/instances.e2e.ts b/spinifex/runtime/roles/spinifexui/frontend/e2e/instances.e2e.ts index 1f3a45267..85e5fd2af 100644 --- a/spinifex/runtime/roles/spinifexui/frontend/e2e/instances.e2e.ts +++ b/spinifex/runtime/roles/spinifexui/frontend/e2e/instances.e2e.ts @@ -508,3 +508,84 @@ test("an empty account is offered a first launch", async ({ page }) => { await page.getByRole("link", { name: "Launch your first instance" }).click() await expect(page).toHaveURL(/\/ec2\/describe-instances\?create=true$/u) }) + +// The launch form decides whether the cluster has compute from the capacity +// listing alone, so it says "no compute" for an empty list whatever the +// reason the list is empty. An idle production cluster hit exactly that: the +// node replies were dropped and the empty result read as no capacity. +const NO_COMPUTE = + "No compute available. No new instances can be created until compute is available." + +function instanceTypeSet(types: readonly [string, number, number][]): string { + const items = types.map( + ([name, vcpu, memMiB]) => + `${name}` + + `${vcpu}` + + `${memMiB}` + + `x86_64` + + ``, + ) + + return `${items.join("")}` +} + +test("the launch form offers the types the capacity listing reports", async ({ + api, + page, +}) => { + api.handle( + "DescribeInstanceTypes", + ec2( + "DescribeInstanceTypes", + // Three free slots of one type, as the gateway expands a per-type count. + instanceTypeSet([ + ["t3.micro", 2, 1024], + ["t3.micro", 2, 1024], + ["t3.micro", 2, 1024], + ]), + ), + ) + + await page.goto("/ec2/describe-instances?create=true") + + await expect(page.getByText(NO_COMPUTE)).toHaveCount(0) + await expect( + page.getByRole("button", { name: "Create Instance" }), + ).toBeEnabled() +}) + +test("the launch form reports the free slot count for the selected type", async ({ + api, + page, +}) => { + api.handle( + "DescribeInstanceTypes", + ec2( + "DescribeInstanceTypes", + instanceTypeSet([ + ["t3.micro", 2, 1024], + ["t3.micro", 2, 1024], + ]), + ), + ) + + await page.goto("/ec2/describe-instances?create=true") + await page.locator("#instanceType").click() + await page.getByRole("option").filter({ hasText: "t3.micro" }).first().click() + + await expect( + page.getByText("Available capacity for t3.micro: 2"), + ).toBeVisible() +}) + +test("an account with no capacity is told so, and cannot launch", async ({ + page, +}) => { + // The default handler answers with an empty instanceTypeSet. + await page.goto("/ec2/describe-instances?create=true") + + await expect(page.getByText(NO_COMPUTE)).toBeVisible() + await expect( + page.getByRole("button", { name: "Create Instance" }), + ).toBeDisabled() +}) diff --git a/spinifex/runtime/roles/spinifexui/frontend/e2e/live/console.live.ts b/spinifex/runtime/roles/spinifexui/frontend/e2e/live/console.live.ts new file mode 100644 index 000000000..e57747fa8 --- /dev/null +++ b/spinifex/runtime/roles/spinifexui/frontend/e2e/live/console.live.ts @@ -0,0 +1,208 @@ +import type { Page } from "@playwright/test" + +import { expect, liveConfig, test } from "./live" + +const RUNNING_TIMEOUT = 5 * 60_000 +const TERMINATED_TIMEOUT = 5 * 60_000 +const GRAPH_TIMEOUT = 60_000 +const APPEAR_TIMEOUT = 60_000 +const MINIMUM_GRAPHS = 3 + +// The banner the console shows when the capacity listing comes back empty. +const NO_COMPUTE = + "No compute available. No new instances can be created until compute is available." + +// The launch form opens in the instance list's side panel, so the table and +// the form are on one page and a launch can be read from the list directly. +const LAUNCH_PANEL = "/ec2/describe-instances?create=true" + +async function openLaunchForm(page: Page): Promise { + await page.goto(LAUNCH_PANEL) + await expect( + page.getByRole("button", { name: "Create Instance" }), + ).toBeVisible({ timeout: APPEAR_TIMEOUT }) +} + +async function instanceIds(page: Page): Promise { + const cells = await page + .getByRole("table", { name: "Instances" }) + .getByRole("cell") + .filter({ hasText: /^i-[0-9a-f]+$/ }) + .allInnerTexts() + + return cells.map((text) => text.trim()) +} + +async function selectOption( + page: Page, + trigger: string, + match?: RegExp, +): Promise { + await page.locator(trigger).click() + const options = page.getByRole("option") + const wanted = match ? options.filter({ hasText: match }) : options + await wanted.first().click() +} + +async function terminate(page: Page, instanceId: string): Promise { + await page.goto("/ec2/describe-instances") + await page.getByRole("checkbox", { name: `Select ${instanceId}` }).check() + await page.getByRole("button", { name: "Terminate", exact: true }).click() + + // The dialog holds its confirm disabled until the phrase is typed, as AWS + // does before an irreversible delete. + await page + .getByRole("textbox", { name: /type .terminate./ }) + .fill("terminate") + await page.getByRole("button", { name: /^Terminate \d+$/ }).click() +} + +// Signing in with an access key is the only way into the console, so every +// other check depends on this one. +test("an operator signs in with an access key and reaches the dashboard", async ({ + signedIn, +}) => { + await signedIn.goto("/") + await expect(signedIn.getByRole("link", { name: "Dashboard" })).toBeVisible({ + timeout: APPEAR_TIMEOUT, + }) + + // The nodes page is where the cluster's own capacity is reported, from the + // daemon's node status rather than from the EC2 listing. + await signedIn.goto("/nodes") + await expect( + signedIn.getByRole("table", { name: "Resource usage by node" }), + ).toBeVisible({ timeout: APPEAR_TIMEOUT }) + await expect( + signedIn.getByRole("table", { name: "Instance type capacity" }), + ).toBeVisible() +}) + +// The regression guard for the defect this suite was written for: a cluster +// with free capacity must not report that it has none. The mocked suite +// cannot see this, because the fault was in what the control plane sent. +test("a cluster with free capacity offers instance types rather than reporting none", async ({ + signedIn, +}) => { + await openLaunchForm(signedIn) + + await expect(signedIn.getByText(NO_COMPUTE)).toHaveCount(0) + await expect( + signedIn.getByRole("button", { name: "Create Instance" }), + ).toBeEnabled() + + await signedIn.locator("#instanceType").click() + await expect(signedIn.getByRole("option").first()).toBeVisible() +}) + +// The console reports the cluster's true free capacity, so a figure capped to +// keep a response small would fail here rather than mislead a customer. +test("the launch form reports a capacity figure for the selected type", async ({ + signedIn, +}) => { + const config = liveConfig() + await openLaunchForm(signedIn) + await selectOption(signedIn, "#instanceType", new RegExp(config.instanceType)) + + const capacity = signedIn.getByText( + new RegExp(`Available capacity for ${config.instanceType}: \\d+`), + ) + + await expect(capacity).toBeVisible() + + const text = (await capacity.textContent()) ?? "" + const match = /: (?\d+)/.exec(text) + expect(Number(match?.groups?.slots ?? 0)).toBeGreaterThan(0) +}) + +// The whole customer path in one test, because the parts are only worth +// anything together: launch, reach running, read live metrics, terminate. +test("an instance launches, serves metrics and terminates", async ({ + signedIn, +}) => { + const config = liveConfig() + await openLaunchForm(signedIn) + const before = new Set(await instanceIds(signedIn)) + let launched = "" + + try { + // The OS tiles put a visually-hidden radio inside its label, so the click + // goes to the label and the browser forwards it. The first tile picks a + // distribution at its newest version for the architecture. + await signedIn.locator('label:has(input[type="radio"])').first().click() + await selectOption( + signedIn, + "#instanceType", + new RegExp(config.instanceType), + ) + await selectOption(signedIn, "#keyName") + await signedIn.getByRole("button", { name: "Create Instance" }).click() + + await expect + .poll( + async () => { + const now = await instanceIds(signedIn) + launched = now.find((id) => !before.has(id)) ?? "" + + return launched + }, + { timeout: APPEAR_TIMEOUT }, + ) + .not.toBe("") + + const row = signedIn.getByRole("row").filter({ hasText: launched }) + await expect(row.getByText("running")).toBeVisible({ + timeout: RUNNING_TIMEOUT, + }) + + // Monitoring is the only page that reads the metrics stream, so a graph + // rendering is the evidence the guest is being sampled. + await signedIn.goto(`/ec2/describe-instances/${launched}?tab=monitoring`) + await expect( + signedIn.getByRole("region", { name: "Instance details" }), + ).toBeVisible() + await expect + .poll(async () => await signedIn.getByRole("figure").count(), { + timeout: GRAPH_TIMEOUT, + }) + .toBeGreaterThanOrEqual(MINIMUM_GRAPHS) + await expect( + signedIn.getByRole("figure", { name: /CPU utilization/i }), + ).toBeVisible() + } finally { + if (launched) { + await terminate(signedIn, launched) + await expect( + signedIn + .getByRole("row") + .filter({ hasText: launched }) + .getByText(/terminated|shutting-down/), + ).toBeVisible({ timeout: TERMINATED_TIMEOUT }) + } + } +}) + +// The pages a customer reaches from the sidebar, each loading its own listing +// against the live control plane. A page that throws renders no table. +const LISTINGS = [ + ["volumes", "/ec2/describe-volumes", "Volumes"], + ["VPCs", "/ec2/describe-vpcs", "VPCs"], + ["subnets", "/ec2/describe-subnets", "Subnets"], + ["security groups", "/ec2/describe-security-groups", "Security groups"], + ["key pairs", "/ec2/describe-key-pairs", "Key pairs"], + ["images", "/ec2/describe-images", "Images"], + ["elastic IPs", "/ec2/describe-addresses", "Elastic IPs"], + ["snapshots", "/ec2/describe-snapshots", "Snapshots"], + ["load balancers", "/ec2/describe-load-balancers", "Load balancers"], +] as const + +for (const [name, path, table] of LISTINGS) { + test(`the ${name} listing loads against the cluster`, async ({ + signedIn, + }) => { + await signedIn.goto(path) + await expect(signedIn.getByRole("table", { name: table })).toBeVisible({ + timeout: APPEAR_TIMEOUT, + }) + }) +} diff --git a/spinifex/runtime/roles/spinifexui/frontend/e2e/live/live.ts b/spinifex/runtime/roles/spinifexui/frontend/e2e/live/live.ts new file mode 100644 index 000000000..a93a32208 --- /dev/null +++ b/spinifex/runtime/roles/spinifexui/frontend/e2e/live/live.ts @@ -0,0 +1,53 @@ +import { test as base, expect, type Page } from "@playwright/test" + +export interface LiveConfig { + consoleUrl: string + accessKeyId: string + secretAccessKey: string + instanceType: string +} + +function required(name: string): string { + const value = process.env[name] + + if (!value) { + throw new Error( + `${name} is not set: the live console suite needs a cluster to run against`, + ) + } + + return value +} + +export function liveConfig(): LiveConfig { + return { + consoleUrl: required("SPX_CONSOLE_URL"), + accessKeyId: required("SPX_ACCESS_KEY_ID"), + secretAccessKey: required("SPX_SECRET_ACCESS_KEY"), + instanceType: process.env.SPX_INSTANCE_TYPE ?? "t3.micro", + } +} + +// Signs in with a long-lived access key, as an operator does. The credentials +// are read from the environment and never written to disk. +export async function signIn(page: Page): Promise { + const config = liveConfig() + + await page.goto("/login") + await page.locator("#accessKeyId").fill(config.accessKeyId) + await page.locator("#secretAccessKey").fill(config.secretAccessKey) + await page.getByRole("button", { name: "Sign In" }).click() + + // The console routes away from /login once the session is stored. + await expect(page).not.toHaveURL(/\/login/, { timeout: 30_000 }) +} + +export const test = base.extend<{ signedIn: Page }>({ + // The second argument is Playwright's fixture callback, not React's use(). + signedIn: async ({ page }, provide) => { + await signIn(page) + await provide(page) + }, +}) + +export { expect } from "@playwright/test" diff --git a/spinifex/runtime/roles/spinifexui/frontend/package.json b/spinifex/runtime/roles/spinifexui/frontend/package.json index 04b34a233..f2f90e377 100644 --- a/spinifex/runtime/roles/spinifexui/frontend/package.json +++ b/spinifex/runtime/roles/spinifexui/frontend/package.json @@ -12,7 +12,8 @@ "test:coverage": "vitest run --coverage", "test:watch": "vitest", "e2e": "playwright test", - "e2e:update": "playwright test --update-snapshots" + "e2e:update": "playwright test --update-snapshots", + "e2e:live": "playwright test --config playwright.live.config.ts" }, "dependencies": { "@aws-crypto/sha256-browser": "^5.2.0", diff --git a/spinifex/runtime/roles/spinifexui/frontend/playwright.live.config.ts b/spinifex/runtime/roles/spinifexui/frontend/playwright.live.config.ts new file mode 100644 index 000000000..9ca3be2d7 --- /dev/null +++ b/spinifex/runtime/roles/spinifexui/frontend/playwright.live.config.ts @@ -0,0 +1,37 @@ +import { defineConfig, devices } from "@playwright/test" + +// The console served by a real cluster, with no API interception. The mocked +// suite in playwright.config.ts cannot see a defect that lives in what the +// control plane actually sends, which is how an idle production cluster came +// to report no compute available. +// +// Pointed at a cluster by SPX_CONSOLE_URL, with SPX_ACCESS_KEY_ID and +// SPX_SECRET_ACCESS_KEY for an account that may launch instances. It creates +// and terminates real resources, so it never runs by default and starts no +// server of its own. +const CONSOLE_URL = process.env.SPX_CONSOLE_URL + +export default defineConfig({ + testDir: "e2e/live", + testMatch: "**/*.live.ts", + // Serial: the suite launches guests and a parallel run would race the + // cluster's capacity against itself. + workers: 1, + timeout: 10 * 60_000, + globalTimeout: 40 * 60_000, + forbidOnly: Boolean(process.env.CI), + reporter: process.env.CI ? "line" : "list", + retries: 0, + use: { + ...devices["Desktop Chrome"], + baseURL: CONSOLE_URL, + // The cluster CA is its own, and the console is reached by node name. + ignoreHTTPSErrors: true, + reducedMotion: "reduce", + locale: "en-AU", + timezoneId: "UTC", + trace: "retain-on-failure", + screenshot: "only-on-failure", + video: "retain-on-failure", + }, +})