mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-23 20:57:02 +00:00
fix: report InsufficientInstanceCapacity instead of a raw NATS no-responders error when RunInstances finds no launch subscriber on a node (#1235)
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user