diff --git a/spinifex/domains/ec2/awsapi/instance/placement.go b/spinifex/domains/ec2/awsapi/instance/placement.go index 63d45547a..0d5dabf90 100644 --- a/spinifex/domains/ec2/awsapi/instance/placement.go +++ b/spinifex/domains/ec2/awsapi/instance/placement.go @@ -227,7 +227,7 @@ func aggregateResults(ctx context.Context, results []nodeLaunchResult, minCount return nil, clientErr } // A launch that was attempted and failed is not a capacity shortage; surface the real cause. - if failErr := launchFailure(results); failErr != nil { + if failErr := launchFailure(ctx, results); failErr != nil { slog.ErrorContext(ctx, "distributeInstances: launch fell short of minCount", "minCount", minCount, "launched", totalLaunched, "err", failErr) return nil, failErr } @@ -321,7 +321,7 @@ func distributeInstancesSpread(ctx context.Context, input *ec2.RunInstancesInput if clientErr := extractClientError(results); clientErr != nil { return nil, clientErr } - if failErr := launchFailure(results); failErr != nil { + if failErr := launchFailure(ctx, results); failErr != nil { slog.ErrorContext(ctx, "distributeInstancesSpread: launch fell short of minCount", "minCount", minCount, "launched", totalLaunched, "err", failErr) return nil, failErr } @@ -414,7 +414,7 @@ func distributeInstancesCluster(ctx context.Context, input *ec2.RunInstancesInpu if clientErr := extractClientError(results); clientErr != nil { return nil, clientErr } - return nil, results[0].Err + return nil, launchFailure(ctx, results) } reservation := results[0].Reservation @@ -468,17 +468,26 @@ func extractClientError(results []nodeLaunchResult) error { return nil } -// launchFailure returns the first node launch error, or nil when no node errored. -// A sub-minCount launch whose nodes actually errored is not a capacity shortage: -// genuine capacity is caught pre-launch in distributeInstances, and a node-side -// capacity race is surfaced by extractClientError. Surfacing the real error (RPC -// timeout, network/OVN failure) stops it masquerading as -// InsufficientInstanceCapacity — the mislabel that derails triage. -func launchFailure(results []nodeLaunchResult) error { +// launchFailure returns the first real node launch error so an RPC or OVN failure is +// not mislabelled as capacity; a node with no launch subscriber for the type cannot +// place it, so when only those failed the result is InsufficientInstanceCapacity. +func launchFailure(ctx context.Context, results []nodeLaunchResult) error { + var noResponders error for _, r := range results { - if r.Err != nil { - return r.Err + if r.Err == nil { + continue } + if errors.Is(r.Err, nats.ErrNoResponders) { + if noResponders == nil { + noResponders = r.Err + } + continue + } + return r.Err + } + if noResponders != nil { + slog.WarnContext(ctx, "distributeInstances: no launch subscriber on node, reporting InsufficientInstanceCapacity", "err", noResponders) + return errors.New(awserrors.ErrorInsufficientInstanceCapacity) } return nil } diff --git a/spinifex/domains/ec2/awsapi/instance/placement_test.go b/spinifex/domains/ec2/awsapi/instance/placement_test.go index 4e4c09714..4a8c7b409 100644 --- a/spinifex/domains/ec2/awsapi/instance/placement_test.go +++ b/spinifex/domains/ec2/awsapi/instance/placement_test.go @@ -335,6 +335,32 @@ func TestAggregateResults_TimeoutSurfacedNotMasked(t *testing.T) { assert.ErrorIs(t, err, context.DeadlineExceeded) } +// A node with no subscriber on its launch subject surfaces as capacity, never the raw NATS error. +func TestAggregateResults_NoRespondersIsCapacity(t *testing.T) { + t.Parallel() + noResp := fmt.Errorf("launch on node-1: %w", fmt.Errorf("NATS request to ec2.RunInstances.t3.large.node-1: %w", nats.ErrNoResponders)) + results := []nodeLaunchResult{ + {NodeID: "node-1", Err: noResp}, + } + + _, err := aggregateResults(context.Background(), results, 1, nil, "") + require.EqualError(t, err, awserrors.ErrorInsufficientInstanceCapacity) +} + +// A real launch failure on another node outranks a no-responders node. +func TestAggregateResults_RealFailureOutranksNoResponders(t *testing.T) { + t.Parallel() + noResp := fmt.Errorf("launch on node-1: %w", nats.ErrNoResponders) + timeoutErr := fmt.Errorf("launch on node-2: %w", context.DeadlineExceeded) + results := []nodeLaunchResult{ + {NodeID: "node-1", Err: noResp}, + {NodeID: "node-2", Err: timeoutErr}, + } + + _, err := aggregateResults(context.Background(), results, 1, nil, "") + assert.ErrorIs(t, err, context.DeadlineExceeded) +} + // Short launch with no node errors (nodes returned fewer instances than assigned // without erroring) keeps the generic capacity code. func TestAggregateResults_ShortWithoutNodeErrors(t *testing.T) { @@ -1120,6 +1146,40 @@ func TestDistributeInstancesCluster_InsufficientCapacityOnPinnedNode(t *testing. assert.Equal(t, awserrors.ErrorInsufficientInstanceCapacity, err.Error()) } +// A pinned node with no launch subscriber reports capacity, not the raw NATS error. +func TestDistributeInstancesCluster_NoLaunchSubscriberIsCapacity(t *testing.T) { + t.Parallel() + _, nc := startTestNATSServer(t) + + statusSub, err := nc.Subscribe("spinifex.node.status", func(msg *nats.Msg) { + data, _ := json.Marshal(types.NodeStatusResponse{Node: "node-1", InstanceTypes: []types.InstanceTypeCap{{Name: "t3.micro", Available: 10}}}) + _ = nc.Publish(msg.Reply, data) + }) + require.NoError(t, err) + defer statusSub.Unsubscribe() + + reserveSub, err := nc.QueueSubscribe("ec2.ReserveClusterNode", "spinifex-workers", func(msg *nats.Msg) { + data, _ := json.Marshal(struct { + TargetNode string `json:"target_node"` + }{TargetNode: "node-1"}) + _ = msg.Respond(data) + }) + require.NoError(t, err) + defer reserveSub.Unsubscribe() + + require.NoError(t, nc.Flush()) + + input := &ec2.RunInstancesInput{ + ImageId: aws.String("ami-test"), + InstanceType: aws.String("t3.micro"), + MinCount: aws.Int64(1), + MaxCount: aws.Int64(1), + } + + _, err = distributeInstancesCluster(context.Background(), input, nc, "test-account", "my-cluster-group", 1) + require.EqualError(t, err, awserrors.ErrorInsufficientInstanceCapacity) +} + func TestDistributeInstancesCluster_PinnedNodeNotInCapacityResults(t *testing.T) { t.Parallel() // Pinned node has no capacity at all (not in fan-out results) → InsufficientInstanceCapacity