From 10ce0fba8ffc38c6cd105016ecc3ba63137d0971 Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Wed, 22 Jul 2026 11:36:59 +1000 Subject: [PATCH] fix: resolve wrapped AWS error codes (#560) --- spinifex/awserrors/awserrors.go | 32 +++++++++++++++ spinifex/awserrors/awserrors_test.go | 41 +++++++++++++++++++ spinifex/daemon/daemon_handlers.go | 10 ++--- .../daemon_handlers_capacityreservation.go | 2 +- spinifex/daemon/daemon_handlers_eni.go | 15 ++----- .../daemon/daemon_handlers_iam_profile.go | 2 +- spinifex/daemon/daemon_handlers_image.go | 2 +- spinifex/daemon/daemon_handlers_instance.go | 6 +-- .../daemon/daemon_handlers_spotinstance.go | 2 +- spinifex/daemon/daemon_handlers_test.go | 25 +++++++++++ spinifex/daemon/daemon_handlers_volume.go | 2 +- spinifex/gateway/ec2/instance/RunInstances.go | 2 +- spinifex/gateway/ec2/instance/placement.go | 8 ++-- .../gateway/ec2/instance/placement_test.go | 21 ++++++++++ spinifex/gateway/gateway.go | 20 ++++----- spinifex/gateway/gateway_test.go | 21 ++++++++++ .../handlers/ec2/instance/service_impl.go | 7 +--- .../ec2/instance/service_impl_test.go | 17 ++++++++ spinifex/handlers/ec2/vpc/external_ipam.go | 3 +- .../handlers/ec2/vpc/external_ipam_test.go | 6 ++- spinifex/handlers/eks/nodegroup.go | 2 +- spinifex/handlers/eks/nodegroup_test.go | 10 +++++ spinifex/network/external/static_pool.go | 3 +- spinifex/network/external/static_pool_test.go | 6 ++- spinifex/services/awsgw/eks_token_verify.go | 2 +- spinifex/utils/nats.go | 19 ++++++++- spinifex/utils/nats_test.go | 35 +++++++++++++++- 27 files changed, 265 insertions(+), 56 deletions(-) diff --git a/spinifex/awserrors/awserrors.go b/spinifex/awserrors/awserrors.go index aa188812a..3b6978c88 100644 --- a/spinifex/awserrors/awserrors.go +++ b/spinifex/awserrors/awserrors.go @@ -522,6 +522,38 @@ func ValidErrorCode(code string) string { return ErrorServerInternal } +// ResolveErrorCode returns the first registered AWS error code in err's unwrap tree. +func ResolveErrorCode(err error) (string, bool) { + if err == nil { + return "", false + } + code := err.Error() + if _, ok := ErrorLookup[code]; ok { + return code, true + } + + if joined, ok := err.(interface{ Unwrap() []error }); ok { + for _, inner := range joined.Unwrap() { + if code, found := ResolveErrorCode(inner); found { + return code, true + } + } + return "", false + } + if wrapped, ok := err.(interface{ Unwrap() error }); ok { + return ResolveErrorCode(wrapped.Unwrap()) + } + return "", false +} + +// ValidErrorCodeFromError resolves err or returns ErrorServerInternal. +func ValidErrorCodeFromError(err error) string { + if code, ok := ResolveErrorCode(err); ok { + return code + } + return ErrorServerInternal +} + // HasErrorCode reports whether s is exactly a registered AWS error code. // Use it to decide whether an error string may be surfaced to clients verbatim // (mapped to its real HTTP status) rather than wrapped/sanitized to 500. diff --git a/spinifex/awserrors/awserrors_test.go b/spinifex/awserrors/awserrors_test.go index e7d9cef96..fcdacf404 100644 --- a/spinifex/awserrors/awserrors_test.go +++ b/spinifex/awserrors/awserrors_test.go @@ -70,6 +70,47 @@ func TestValidErrorCode(t *testing.T) { } } +func TestResolveErrorCode(t *testing.T) { + known := errors.New(ErrorInsufficientAddressCapacity) + tests := []struct { + name string + err error + want string + wantOK bool + }{ + {name: "nil", err: nil}, + {name: "exact code", err: known, want: ErrorInsufficientAddressCapacity, wantOK: true}, + {name: "single wrapper", err: fmt.Errorf("allocate public address: %w", known), want: ErrorInsufficientAddressCapacity, wantOK: true}, + {name: "nested wrappers", err: fmt.Errorf("launch on node: %w", fmt.Errorf("prepare instance: %w", known)), want: ErrorInsufficientAddressCapacity, wantOK: true}, + {name: "joined errors", err: errors.Join(errors.New("opaque"), known), want: ErrorInsufficientAddressCapacity, wantOK: true}, + {name: "first joined code wins", err: errors.Join(errors.New(ErrorAuthFailure), known), want: ErrorAuthFailure, wantOK: true}, + {name: "unknown error", err: errors.New("opaque")}, + {name: "code substring is not exact", err: errors.New("launch failed: " + ErrorInsufficientAddressCapacity)}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, ok := ResolveErrorCode(tt.err) + if got != tt.want || ok != tt.wantOK { + t.Errorf("ResolveErrorCode() = (%q, %v), want (%q, %v)", got, ok, tt.want, tt.wantOK) + } + }) + } +} + +func TestValidErrorCodeFromError(t *testing.T) { + wrapped := fmt.Errorf("launch: %w", errors.New(ErrorInsufficientInstanceCapacity)) + if got := ValidErrorCodeFromError(wrapped); got != ErrorInsufficientInstanceCapacity { + t.Errorf("ValidErrorCodeFromError() = %q, want %q", got, ErrorInsufficientInstanceCapacity) + } + if got := ValidErrorCodeFromError(errors.New("opaque")); got != ErrorServerInternal { + t.Errorf("ValidErrorCodeFromError() = %q, want %q", got, ErrorServerInternal) + } + if got := ValidErrorCodeFromError(nil); got != ErrorServerInternal { + t.Errorf("ValidErrorCodeFromError(nil) = %q, want %q", got, ErrorServerInternal) + } +} + func TestHasErrorCode(t *testing.T) { tests := []struct { name string diff --git a/spinifex/daemon/daemon_handlers.go b/spinifex/daemon/daemon_handlers.go index 259a1e95e..c1fc6fd2c 100644 --- a/spinifex/daemon/daemon_handlers.go +++ b/spinifex/daemon/daemon_handlers.go @@ -91,7 +91,7 @@ func handleNATSRequest[I any, O any](serviceFn func(context.Context, *I, string) // in the journal too. slog.ErrorContext(ctx, "handleNATSRequest: service call failed", "subject", msg.Subject, "err", err) utils.MarkSpanError(span, err) - respondWithError(msg, awserrors.ValidErrorCode(err.Error())) + respondWithError(msg, awserrors.ValidErrorCodeFromError(err)) return } respondWithJSON(msg, output) @@ -120,7 +120,7 @@ func handleNATSRequestWithPrincipal[I any, O any](serviceFn func(context.Context output, err := serviceFn(ctx, input, accountID, principalARN) if err != nil { utils.MarkSpanError(span, err) - respondWithError(msg, awserrors.ValidErrorCode(err.Error())) + respondWithError(msg, awserrors.ValidErrorCodeFromError(err)) return } respondWithJSON(msg, output) @@ -176,7 +176,7 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) { endOpSpan(opSpan, err) if err != nil { utils.MarkSpanError(span, err) - respondWithError(msg, awserrors.ValidErrorCode(err.Error())) + respondWithError(msg, awserrors.ValidErrorCodeFromError(err)) return } if err := msg.Respond(fmt.Appendf(nil, `{"status":"running","instanceId":"%s"}`, instance.ID)); err != nil { @@ -188,7 +188,7 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) { endOpSpan(opSpan, err) if err != nil { utils.MarkSpanError(span, err) - respondWithError(msg, awserrors.ValidErrorCode(err.Error())) + respondWithError(msg, awserrors.ValidErrorCodeFromError(err)) return } if err := msg.Respond([]byte(`{}`)); err != nil { @@ -204,7 +204,7 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) { endOpSpan(opSpan, err) if err != nil { utils.MarkSpanError(span, err) - respondWithError(msg, awserrors.ValidErrorCode(err.Error())) + respondWithError(msg, awserrors.ValidErrorCodeFromError(err)) return } if err := msg.Respond([]byte(`{}`)); err != nil { diff --git a/spinifex/daemon/daemon_handlers_capacityreservation.go b/spinifex/daemon/daemon_handlers_capacityreservation.go index 1543ec374..7e6b77126 100644 --- a/spinifex/daemon/daemon_handlers_capacityreservation.go +++ b/spinifex/daemon/daemon_handlers_capacityreservation.go @@ -58,7 +58,7 @@ func (d *Daemon) handleEC2CreateCapacityReservation(msg *nats.Msg) { } if err := d.resourceMgr.CreateReservation(rec); err != nil { - respondWithError(msg, awserrors.ValidErrorCode(err.Error())) + respondWithError(msg, awserrors.ValidErrorCodeFromError(err)) return } diff --git a/spinifex/daemon/daemon_handlers_eni.go b/spinifex/daemon/daemon_handlers_eni.go index 1baf248b0..2cb7f1213 100644 --- a/spinifex/daemon/daemon_handlers_eni.go +++ b/spinifex/daemon/daemon_handlers_eni.go @@ -39,7 +39,7 @@ func (d *Daemon) handleAttachNetworkInterface(ctx context.Context, msg *nats.Msg record, err := d.vpcService.GetENIRecord(accountID, eniID) if err != nil { - respondWithError(msg, errorCodeFor(err)) + respondWithError(msg, awserrors.ValidErrorCodeFromError(err)) return } if record.Status == "in-use" && record.InstanceId != instance.ID { @@ -49,7 +49,7 @@ func (d *Daemon) handleAttachNetworkInterface(ctx context.Context, msg *nats.Msg attachmentID, err := d.vpcService.AttachENI(accountID, eniID, instance.ID, deviceIndex) if err != nil { - respondWithError(msg, errorCodeFor(err)) + respondWithError(msg, awserrors.ValidErrorCodeFromError(err)) return } if err := d.vpcService.UpdateENI(accountID, eniID, func(r *handlers_ec2_vpc.ENIRecord) { @@ -117,7 +117,7 @@ func (d *Daemon) handleDetachNetworkInterface(ctx context.Context, msg *nats.Msg record, err := d.vpcService.FindENIByAttachment(accountID, attachmentID) if err != nil { - respondWithError(msg, errorCodeFor(err)) + respondWithError(msg, awserrors.ValidErrorCodeFromError(err)) return } if record.InstanceId != instance.ID { @@ -191,15 +191,6 @@ func eniHotplugErrorCode(err error) string { } } -// errorCodeFor passes through errors whose message is already a valid AWS -// code (the convention VPCServiceImpl uses for its sentinel errors). -func errorCodeFor(err error) string { - if err == nil { - return "" - } - return awserrors.ValidErrorCode(err.Error()) -} - // publishENIHotplugEvent emits the per-instance NATS event on best-effort // basis — observers (Phase 4 ecs-agent, future vpcd subscribers) read it // for fanout. A publish failure does not roll back the pipeline. diff --git a/spinifex/daemon/daemon_handlers_iam_profile.go b/spinifex/daemon/daemon_handlers_iam_profile.go index d451d2221..5dd868f11 100644 --- a/spinifex/daemon/daemon_handlers_iam_profile.go +++ b/spinifex/daemon/daemon_handlers_iam_profile.go @@ -15,7 +15,7 @@ import ( func (d *Daemon) handleAssociateIamInstanceProfile(ctx context.Context, msg *nats.Msg, command types.EC2InstanceCommand, instance *vm.VM) { result, err := d.instanceService.AssociateIamInstanceProfile(ctx, instance, command) if err != nil { - respondWithError(msg, awserrors.ValidErrorCode(err.Error())) + respondWithError(msg, awserrors.ValidErrorCodeFromError(err)) return } respondWithJSON(msg, result) diff --git a/spinifex/daemon/daemon_handlers_image.go b/spinifex/daemon/daemon_handlers_image.go index fbd8a7efe..251801b5e 100644 --- a/spinifex/daemon/daemon_handlers_image.go +++ b/spinifex/daemon/daemon_handlers_image.go @@ -130,7 +130,7 @@ func (d *Daemon) handleEC2CreateImage(msg *nats.Msg) { output, err := d.imageService.CreateImageFromInstance(params, accountID) if err != nil { slog.Error("CreateImage: service failed", "instanceId", instanceID, "err", err) - respondWithError(msg, awserrors.ValidErrorCode(err.Error())) + respondWithError(msg, awserrors.ValidErrorCodeFromError(err)) return } diff --git a/spinifex/daemon/daemon_handlers_instance.go b/spinifex/daemon/daemon_handlers_instance.go index 00a3a0167..6cf0a50f8 100644 --- a/spinifex/daemon/daemon_handlers_instance.go +++ b/spinifex/daemon/daemon_handlers_instance.go @@ -71,7 +71,7 @@ func (d *Daemon) handleSetInstanceTags(ctx context.Context, msg *nats.Msg, comma return true }) if err != nil { - respondWithError(msg, awserrors.ValidErrorCode(err.Error())) + respondWithError(msg, awserrors.ValidErrorCodeFromError(err)) return } if !found { @@ -119,7 +119,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { 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())) + respondWithError(msg, awserrors.ValidErrorCodeFromError(vErr)) return } } @@ -129,7 +129,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { reservation, instances, instanceType, err := d.instanceService.PrepareRunInstances(ctx, input, accountID, reservationID) endOpSpan(prepSpan, err) if err != nil { - respondWithError(msg, awserrors.ValidErrorCode(err.Error())) + respondWithError(msg, awserrors.ValidErrorCodeFromError(err)) return } diff --git a/spinifex/daemon/daemon_handlers_spotinstance.go b/spinifex/daemon/daemon_handlers_spotinstance.go index 867897787..356540f8d 100644 --- a/spinifex/daemon/daemon_handlers_spotinstance.go +++ b/spinifex/daemon/daemon_handlers_spotinstance.go @@ -27,7 +27,7 @@ func (d *Daemon) handleSetSpotLineage(ctx context.Context, msg *nats.Msg, comman return true }) if err != nil { - respondWithError(msg, awserrors.ValidErrorCode(err.Error())) + respondWithError(msg, awserrors.ValidErrorCodeFromError(err)) return } if !found { diff --git a/spinifex/daemon/daemon_handlers_test.go b/spinifex/daemon/daemon_handlers_test.go index 9caeda68e..f5f538f3c 100644 --- a/spinifex/daemon/daemon_handlers_test.go +++ b/spinifex/daemon/daemon_handlers_test.go @@ -5,6 +5,7 @@ import ( "context" "encoding/base64" "encoding/json" + "errors" "fmt" "log/slog" "os" @@ -222,6 +223,30 @@ func TestHandleNATSRequest_ServiceError(t *testing.T) { assert.Equal(t, awserrors.ErrorServerInternal, errResp["Code"]) } +func TestHandleNATSRequest_WrappedAWSServiceError(t *testing.T) { + nc, err := nats.Connect(sharedNATSURL) + require.NoError(t, err) + defer nc.Close() + + serviceFn := func(_ context.Context, _ *testInput, _ string) (*testOutput, error) { + cause := errors.New(awserrors.ErrorInsufficientAddressCapacity) + return nil, fmt.Errorf("allocate public address: %w", cause) + } + + sub, err := nc.Subscribe("test.err.wrapped", handleNATSRequest(serviceFn)) + require.NoError(t, err) + defer sub.Unsubscribe() + + reqData, err := json.Marshal(testInput{Name: "world"}) + require.NoError(t, err) + reply, err := nc.Request("test.err.wrapped", reqData, 5*time.Second) + require.NoError(t, err) + + var errResp map[string]any + require.NoError(t, json.Unmarshal(reply.Data, &errResp)) + assert.Equal(t, awserrors.ErrorInsufficientAddressCapacity, errResp["Code"]) +} + // --- Handler wrapper tests (representative set via NATS round-trip) --- func TestHandleEC2CreateKeyPair_RoundTrip(t *testing.T) { diff --git a/spinifex/daemon/daemon_handlers_volume.go b/spinifex/daemon/daemon_handlers_volume.go index 89f9982c4..13554f39c 100644 --- a/spinifex/daemon/daemon_handlers_volume.go +++ b/spinifex/daemon/daemon_handlers_volume.go @@ -189,7 +189,7 @@ func (d *Daemon) handleEC2ModifyVolume(msg *nats.Msg) { if err != nil { slog.ErrorContext(ctx, "handleEC2ModifyVolume service.ModifyVolume failed", "err", err) utils.MarkSpanError(span, err) - respondWithError(msg, awserrors.ValidErrorCode(err.Error())) + respondWithError(msg, awserrors.ValidErrorCodeFromError(err)) return } diff --git a/spinifex/gateway/ec2/instance/RunInstances.go b/spinifex/gateway/ec2/instance/RunInstances.go index 8b5f2e8c6..fb6a70a71 100644 --- a/spinifex/gateway/ec2/instance/RunInstances.go +++ b/spinifex/gateway/ec2/instance/RunInstances.go @@ -168,7 +168,7 @@ func runInstancesInner(ctx context.Context, input *ec2.RunInstancesInput, natsCo reservationPtr, err := distributeInstances(ctx, input, natsConn, accountID, expectedNodes) if err != nil { // Distinguish "unknown type" from "no capacity" via DescribeInstanceTypes. - if err.Error() == awserrors.ErrorInsufficientInstanceCapacity { + if code, ok := awserrors.ResolveErrorCode(err); ok && code == awserrors.ErrorInsufficientInstanceCapacity { if !isKnownInstanceType(ctx, natsConn, *input.InstanceType) { return reservation, errors.New(awserrors.ErrorInvalidInstanceType) } diff --git a/spinifex/gateway/ec2/instance/placement.go b/spinifex/gateway/ec2/instance/placement.go index 62d0ca165..f60e6bd5d 100644 --- a/spinifex/gateway/ec2/instance/placement.go +++ b/spinifex/gateway/ec2/instance/placement.go @@ -437,11 +437,11 @@ func extractClientError(results []nodeLaunchResult) error { if r.Err == nil { continue } - inner := errors.Unwrap(r.Err) - if inner == nil { + code, ok := awserrors.ResolveErrorCode(r.Err) + if !ok { continue } - switch inner.Error() { + switch code { case awserrors.ErrorInsufficientInstanceCapacity, awserrors.ErrorInvalidAMIIDNotFound, awserrors.ErrorInvalidAMIIDMalformed, @@ -451,7 +451,7 @@ func extractClientError(results []nodeLaunchResult) error { awserrors.ErrorInvalidGroupNotFound, awserrors.ErrorInvalidParameterValue, awserrors.ErrorSecurityGroupsPerInterfaceLimitExceeded: - return inner + return errors.New(code) } } return nil diff --git a/spinifex/gateway/ec2/instance/placement_test.go b/spinifex/gateway/ec2/instance/placement_test.go index 5f4b831ae..ca21da2de 100644 --- a/spinifex/gateway/ec2/instance/placement_test.go +++ b/spinifex/gateway/ec2/instance/placement_test.go @@ -371,6 +371,27 @@ func TestExtractClientError_AMINotFound(t *testing.T) { assert.Equal(t, awserrors.ErrorInvalidAMIIDNotFound, err.Error()) } +func TestExtractClientError_NestedWrappedCode(t *testing.T) { + inner := errors.New(awserrors.ErrorInvalidParameterValue) + wrapped := fmt.Errorf("launch on node-1: %w", fmt.Errorf("prepare instance: %w", inner)) + results := []nodeLaunchResult{ + {NodeID: "node-1", Err: wrapped}, + } + + err := extractClientError(results) + require.EqualError(t, err, awserrors.ErrorInvalidParameterValue) +} + +func TestExtractClientError_JoinedCode(t *testing.T) { + joined := errors.Join(assert.AnError, errors.New(awserrors.ErrorInvalidGroupNotFound)) + results := []nodeLaunchResult{ + {NodeID: "node-1", Err: fmt.Errorf("launch on node-1: %w", joined)}, + } + + err := extractClientError(results) + require.EqualError(t, err, awserrors.ErrorInvalidGroupNotFound) +} + func TestExtractClientError_KeyPairNotFound(t *testing.T) { inner := errors.New(awserrors.ErrorInvalidKeyPairNotFound) wrapped := fmt.Errorf("launch on node-1: %w", inner) diff --git a/spinifex/gateway/gateway.go b/spinifex/gateway/gateway.go index 7df705e8f..32b10002e 100644 --- a/spinifex/gateway/gateway.go +++ b/spinifex/gateway/gateway.go @@ -550,15 +550,13 @@ func (gw *GatewayConfig) ErrorHandler(w http.ResponseWriter, r *http.Request, er slog.Debug("ErrorHandler", "service", svc, "error", err.Error()) var requestId = uuid.NewString() - var errorMsg = awserrors.ErrorMessage{} - - if _, exists := awserrors.ErrorLookup[err.Error()]; !exists { + code, exists := awserrors.ResolveErrorCode(err) + if !exists { slog.Warn("Unknown error code", "error", err.Error()) - err = errors.New(awserrors.ErrorInternalError) + code = awserrors.ErrorInternalError } - errorMsg = awserrors.ErrorLookup[err.Error()] - + errorMsg := awserrors.ErrorLookup[code] if errorMsg.HTTPCode == 0 { errorMsg.HTTPCode = 500 } @@ -566,8 +564,8 @@ func (gw *GatewayConfig) ErrorHandler(w http.ResponseWriter, r *http.Request, er // EKS, ECR, ACM, ECS, tagging, and bedrock/bedrock-runtime use AWS JSON 1.1; // query/XML services fall through. if svc == "eks" || svc == "ecr" || svc == "acm" || svc == "ecs" || svc == "tagging" || svc == "bedrock" || svc == "bedrock-runtime" { - body := GenerateEKSErrorResponse(err.Error(), errorMsg.Message, requestId) - slog.Debug("Generated JSON error response", "service", svc, "error", err.Error(), "json", string(body), "requestId", requestId) + body := GenerateEKSErrorResponse(code, errorMsg.Message, requestId) + slog.Debug("Generated JSON error response", "service", svc, "error", err, "code", code, "json", string(body), "requestId", requestId) w.Header().Set("Content-Type", eksJSONContentType) w.WriteHeader(errorMsg.HTTPCode) if _, err := w.Write(body); err != nil { @@ -578,12 +576,12 @@ func (gw *GatewayConfig) ErrorHandler(w http.ResponseWriter, r *http.Request, er var xmlError []byte if svc == "iam" || svc == "sts" || svc == "elasticloadbalancing" { - xmlError = GenerateIAMErrorResponse(err.Error(), errorMsg.Message, requestId) + xmlError = GenerateIAMErrorResponse(code, errorMsg.Message, requestId) } else { - xmlError = GenerateEC2ErrorResponse(err.Error(), errorMsg.Message, requestId) + xmlError = GenerateEC2ErrorResponse(code, errorMsg.Message, requestId) } - slog.Debug("Generated error response", "error", err.Error(), "xml", string(xmlError), "requestId", requestId) + slog.Debug("Generated error response", "error", err, "code", code, "xml", string(xmlError), "requestId", requestId) w.Header().Set("Content-Type", "application/xml") w.WriteHeader(errorMsg.HTTPCode) diff --git a/spinifex/gateway/gateway_test.go b/spinifex/gateway/gateway_test.go index 7e731e4bd..8e8a9e966 100644 --- a/spinifex/gateway/gateway_test.go +++ b/spinifex/gateway/gateway_test.go @@ -199,6 +199,27 @@ func TestErrorHandler_UnknownError(t *testing.T) { assert.Contains(t, xmlStr, "") } +func TestErrorHandler_WrappedErrorCode(t *testing.T) { + gw := &GatewayConfig{DisableLogging: true} + + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ctx := context.WithValue(r.Context(), ctxService, "ec2") + r = r.WithContext(ctx) + cause := errors.New(awserrors.ErrorInsufficientAddressCapacity) + gw.ErrorHandler(w, r, fmt.Errorf("launch on node-1: %w", fmt.Errorf("allocate address: %w", cause))) + }) + + req := httptest.NewRequest(http.MethodPost, "/", nil) + resp := doRequest(handler, req) + assert.Equal(t, 503, resp.StatusCode) + + body, _ := io.ReadAll(resp.Body) + xmlStr := string(body) + assert.Contains(t, xmlStr, ""+awserrors.ErrorInsufficientAddressCapacity+"") + assert.Contains(t, xmlStr, awserrors.ErrorLookup[awserrors.ErrorInsufficientAddressCapacity].Message) + assert.NotContains(t, xmlStr, "launch on node-1") +} + func TestErrorHandler_ELBv2Service(t *testing.T) { gw := &GatewayConfig{DisableLogging: true} diff --git a/spinifex/handlers/ec2/instance/service_impl.go b/spinifex/handlers/ec2/instance/service_impl.go index 418fe0625..d557cd11b 100644 --- a/spinifex/handlers/ec2/instance/service_impl.go +++ b/spinifex/handlers/ec2/instance/service_impl.go @@ -748,12 +748,7 @@ func (s *InstanceServiceImpl) PrepareRunInstances(ctx context.Context, input *ec s.resourceMgr.ReleaseToReservation(reservationID, instanceType) } } - errCode := awserrors.ErrorServerInternal - if lastRunErr != nil { - if _, ok := awserrors.ErrorLookup[lastRunErr.Error()]; ok { - errCode = lastRunErr.Error() - } - } + errCode := awserrors.ValidErrorCodeFromError(lastRunErr) slog.ErrorContext(ctx, "PrepareRunInstances: failed to create minimum instances", "created", len(instances), "minCount", minCount, "err", errCode) return nil, nil, nil, errors.New(errCode) } diff --git a/spinifex/handlers/ec2/instance/service_impl_test.go b/spinifex/handlers/ec2/instance/service_impl_test.go index 526cf514f..2761d6315 100644 --- a/spinifex/handlers/ec2/instance/service_impl_test.go +++ b/spinifex/handlers/ec2/instance/service_impl_test.go @@ -3154,6 +3154,23 @@ func TestPrepareRunInstances_ENICreateFailureDeallocates(t *testing.T) { require.Len(t, prov.deallocated, 1, "ENI failure must trigger deallocate") } +func TestPrepareRunInstances_WrappedENICreateErrorPreservesCode(t *testing.T) { + cause := errors.New(awserrors.ErrorInvalidSubnetIDNotFound) + eni := &fakeENICreator{createErr: fmt.Errorf("create primary ENI: %w", cause)} + svc, prov := prepareSvcWithENI(t, eni, nil) + + _, _, _, err := svc.PrepareRunInstances(context.Background(), &ec2.RunInstancesInput{ + InstanceType: aws.String("t3.micro"), + ImageId: aws.String("ami-1"), + SubnetId: aws.String("subnet-missing"), + MinCount: aws.Int64(1), + MaxCount: aws.Int64(1), + }, "acc", "") + + require.EqualError(t, err, awserrors.ErrorInvalidSubnetIDNotFound) + require.Len(t, prov.deallocated, 1) +} + // TestPrepareRunInstances_ENIAttachFailureRollsBack verifies that an AttachENI // failure deletes the auto-created ENI, deallocates capacity, and drops the // instance from the reservation. diff --git a/spinifex/handlers/ec2/vpc/external_ipam.go b/spinifex/handlers/ec2/vpc/external_ipam.go index 156091c19..faa5e756d 100644 --- a/spinifex/handlers/ec2/vpc/external_ipam.go +++ b/spinifex/handlers/ec2/vpc/external_ipam.go @@ -8,6 +8,7 @@ import ( "net" "net/netip" + "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/network/external" "github.com/mulgadc/spinifex/spinifex/network/external/dhcp" "github.com/nats-io/nats.go" @@ -82,7 +83,7 @@ func (m *ExternalIPAM) EnableDHCP(client *dhcp.NATSClient) error { func (m *ExternalIPAM) AllocateIP(ctx context.Context, region, az, purpose, allocID, eniID, instanceID string) (string, string, error) { pool := m.findPool(region, az) if pool == nil { - return "", "", fmt.Errorf("InsufficientAddressCapacity: no external pool available for region=%q az=%q", region, az) + return "", "", fmt.Errorf("no external pool available for region=%q az=%q: %w", region, az, errors.New(awserrors.ErrorInsufficientAddressCapacity)) } ip, err := m.AllocateFromPool(ctx, pool.Name, purpose, allocID, eniID, instanceID) if err != nil { diff --git a/spinifex/handlers/ec2/vpc/external_ipam_test.go b/spinifex/handlers/ec2/vpc/external_ipam_test.go index f9217c652..24c1799cd 100644 --- a/spinifex/handlers/ec2/vpc/external_ipam_test.go +++ b/spinifex/handlers/ec2/vpc/external_ipam_test.go @@ -5,6 +5,7 @@ import ( "sync" "testing" + "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/network/external" "github.com/mulgadc/spinifex/spinifex/testutil" "github.com/stretchr/testify/assert" @@ -380,7 +381,10 @@ func TestExternalIPAM_NoPoolAvailable(t *testing.T) { // No pool matches eu-west _, _, err := ipam.AllocateIP(context.Background(), "eu-west-1", "eu-west-1a", PurposeENIPublic, "", "eni-1", "i-1") - assert.ErrorContains(t, err, "InsufficientAddressCapacity") + assert.ErrorContains(t, err, `region="eu-west-1" az="eu-west-1a"`) + code, ok := awserrors.ResolveErrorCode(err) + assert.True(t, ok) + assert.Equal(t, awserrors.ErrorInsufficientAddressCapacity, code) } func TestExternalIPAM_FindPoolByName_NotFound(t *testing.T) { diff --git a/spinifex/handlers/eks/nodegroup.go b/spinifex/handlers/eks/nodegroup.go index 2c1071867..a2789b335 100644 --- a/spinifex/handlers/eks/nodegroup.go +++ b/spinifex/handlers/eks/nodegroup.go @@ -931,7 +931,7 @@ func (s *EKSServiceImpl) launchWorkersCAS(ctx context.Context, acctKV nats.KeyVa // Surface a client-facing code (e.g. InsufficientInstanceCapacity) // verbatim so the gateway maps its real status; wrap only opaque // internal failures, which stay 500. - if awserrors.HasErrorCode(err.Error()) { + if _, ok := awserrors.ResolveErrorCode(err); ok { return nil, err } return nil, fmt.Errorf("launch workers: %w", err) diff --git a/spinifex/handlers/eks/nodegroup_test.go b/spinifex/handlers/eks/nodegroup_test.go index d42d56cec..077581438 100644 --- a/spinifex/handlers/eks/nodegroup_test.go +++ b/spinifex/handlers/eks/nodegroup_test.go @@ -3,6 +3,7 @@ package handlers_eks import ( "context" "errors" + "fmt" "slices" "strconv" "strings" @@ -846,6 +847,15 @@ func TestUpdateNodegroupConfig_CapacityErrorSurfacesCode(t *testing.T) { assert.True(t, awserrors.HasErrorCode(err.Error())) }) + t.Run("wrapped capacity code preserved", func(t *testing.T) { + wrapped := fmt.Errorf("launch on node-1: %w", errors.New(awserrors.ErrorInsufficientInstanceCapacity)) + err := scaleUpWithRunErr(t, wrapped) + require.EqualError(t, err, wrapped.Error()) + code, ok := awserrors.ResolveErrorCode(err) + assert.True(t, ok) + assert.Equal(t, awserrors.ErrorInsufficientInstanceCapacity, code) + }) + t.Run("opaque error wrapped", func(t *testing.T) { err := scaleUpWithRunErr(t, errors.New("kvm: out of host memory")) require.Error(t, err) diff --git a/spinifex/network/external/static_pool.go b/spinifex/network/external/static_pool.go index c366bc42d..dd2785dab 100644 --- a/spinifex/network/external/static_pool.go +++ b/spinifex/network/external/static_pool.go @@ -9,6 +9,7 @@ import ( "net" "net/netip" + "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" @@ -339,5 +340,5 @@ func nextAvailableIP(record *PoolRecord) (string, error) { } } - return "", fmt.Errorf("InsufficientAddressCapacity: pool %s exhausted", record.PoolName) + return "", fmt.Errorf("pool %s exhausted: %w", record.PoolName, errors.New(awserrors.ErrorInsufficientAddressCapacity)) } diff --git a/spinifex/network/external/static_pool_test.go b/spinifex/network/external/static_pool_test.go index e0899f935..5ec329013 100644 --- a/spinifex/network/external/static_pool_test.go +++ b/spinifex/network/external/static_pool_test.go @@ -6,6 +6,7 @@ import ( "sync" "testing" + "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/testutil" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -111,7 +112,10 @@ func TestStaticPool_Exhaustion(t *testing.T) { _, err = a.Allocate(ctx, AllocateRequest{PoolName: "tiny"}) require.NoError(t, err) _, err = a.Allocate(ctx, AllocateRequest{PoolName: "tiny"}) - assert.ErrorContains(t, err, "InsufficientAddressCapacity") + assert.ErrorContains(t, err, "pool tiny exhausted") + code, ok := awserrors.ResolveErrorCode(err) + assert.True(t, ok) + assert.Equal(t, awserrors.ErrorInsufficientAddressCapacity, code) } func TestStaticPool_CASConflict(t *testing.T) { diff --git a/spinifex/services/awsgw/eks_token_verify.go b/spinifex/services/awsgw/eks_token_verify.go index 730fc8e9c..cc3d042ba 100644 --- a/spinifex/services/awsgw/eks_token_verify.go +++ b/spinifex/services/awsgw/eks_token_verify.go @@ -48,7 +48,7 @@ func handleEKSTokenVerify(verifier presignedVerifier) nats.MsgHandler { // A verify failure is the common, expected case for a forged/replayed // token — log at debug so a port-scan does not flood the error log. slog.Debug("EKS token verify rejected", "cluster", req.ClusterName, "err", err) - respondTokenVerifyErr(msg, awserrors.ValidErrorCode(err.Error())) + respondTokenVerifyErr(msg, awserrors.ValidErrorCodeFromError(err)) return } diff --git a/spinifex/utils/nats.go b/spinifex/utils/nats.go index d6a35ac35..1723dc2dd 100644 --- a/spinifex/utils/nats.go +++ b/spinifex/utils/nats.go @@ -220,6 +220,21 @@ func PrincipalARNFromMsg(msg *nats.Msg) string { return msg.Header.Get(PrincipalARNHeader) } +// decodedNATSError retains a response's diagnostic message while exposing its +// sanitized AWS wire code to unwrap-aware classifiers. +type decodedNATSError struct { + code error + message string +} + +func (e *decodedNATSError) Error() string { + return e.message +} + +func (e *decodedNATSError) Unwrap() error { + return e.code +} + // NATSRequest performs a NATS request-response with JSON marshaling. It sends // with X-Account-ID (plus any extra headers), unmarshals the successful response // into Out, and carries ctx's trace context onto the wire: it opens a client span @@ -258,7 +273,7 @@ func NATSRequest[Out any](ctx context.Context, conn *nats.Conn, subject string, responseError, err := ValidateErrorPayload(msg.Data) if err != nil { if responseError.Message != nil && *responseError.Message != "" { - return nil, errors.New(*responseError.Message) + return nil, &decodedNATSError{code: errors.New(*responseError.Code), message: *responseError.Message} } return nil, errors.New(*responseError.Code) } @@ -292,7 +307,7 @@ func ServeNATSRequestCtx[I any, O any](msg *nats.Msg, fn func(context.Context, * out, err := fn(ctx, input) if err != nil { MarkSpanError(span, err) - respondNATS(msg, GenerateErrorPayloadWithMessage(awserrors.ValidErrorCode(err.Error()), err.Error())) + respondNATS(msg, GenerateErrorPayloadWithMessage(awserrors.ValidErrorCodeFromError(err), err.Error())) return } data, err := json.Marshal(out) diff --git a/spinifex/utils/nats_test.go b/spinifex/utils/nats_test.go index 71e0345c7..8e665fd3b 100644 --- a/spinifex/utils/nats_test.go +++ b/spinifex/utils/nats_test.go @@ -11,6 +11,8 @@ import ( "crypto/x509/pkix" "encoding/json" "encoding/pem" + "errors" + "fmt" "log/slog" "math/big" "net" @@ -278,7 +280,10 @@ func TestNATSRequest_ErrorResponse(t *testing.T) { type Resp struct{} _, err = NATSRequest[Resp](context.Background(), nc, "test.fail", struct{}{}, 2*time.Second, "") assert.Error(t, err) - assert.Contains(t, err.Error(), "InvalidParameterValue") + assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error()) + code, ok := awserrors.ResolveErrorCode(err) + assert.True(t, ok) + assert.Equal(t, awserrors.ErrorInvalidParameterValue, code) } func TestNATSRequest_ErrorResponseSurfacesMessage(t *testing.T) { @@ -300,6 +305,34 @@ func TestNATSRequest_ErrorResponseSurfacesMessage(t *testing.T) { _, err = NATSRequest[Resp](context.Background(), nc, "test.fail.msg", struct{}{}, 2*time.Second, "") assert.Error(t, err) assert.Equal(t, reason, err.Error()) + code, ok := awserrors.ResolveErrorCode(err) + assert.True(t, ok) + assert.Equal(t, awserrors.ErrorServerInternal, code) +} + +func TestServeNATSRequest_WrappedErrorPreservesCodeAndMessage(t *testing.T) { + ns := startTestNATSServer(t) + + nc, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + defer nc.Close() + + const message = "launch on node-1: InsufficientAddressCapacity" + _, err = nc.Subscribe("test.serve.error", func(msg *nats.Msg) { + ServeNATSRequest(msg, func(_ *struct{}) (*struct{}, error) { + cause := errors.New(awserrors.ErrorInsufficientAddressCapacity) + return nil, fmt.Errorf("launch on node-1: %w", cause) + }) + }) + require.NoError(t, err) + require.NoError(t, nc.Flush()) + + _, err = NATSRequest[struct{}](context.Background(), nc, "test.serve.error", struct{}{}, 2*time.Second, "") + require.Error(t, err) + assert.Equal(t, message, err.Error()) + code, ok := awserrors.ResolveErrorCode(err) + assert.True(t, ok) + assert.Equal(t, awserrors.ErrorInsufficientAddressCapacity, code) } func TestNATSRequest_NoResponders(t *testing.T) {