diff --git a/.github/workflows/release-and-push.yml b/.github/workflows/release-and-push.yml index 2c21e0c..a12a743 100644 --- a/.github/workflows/release-and-push.yml +++ b/.github/workflows/release-and-push.yml @@ -76,6 +76,8 @@ jobs: --label org.opencontainers.image.source=https://github.com/dappnode/dappnode-nexus-gateway \ --label org.opencontainers.image.revision=${{ github.sha }} \ --label org.opencontainers.image.version="$VERSION" \ + --build-arg VERSION="$VERSION" \ + --build-arg COMMIT=${{ github.sha }} \ -t "$ECR_REGISTRY/$IMAGE_NAME:$VERSION" \ -t "$ECR_REGISTRY/$IMAGE_NAME:latest" \ -t "$ECR_REGISTRY/$IMAGE_NAME:sha-${{ github.sha }}" \ diff --git a/apps/gateway/Dockerfile b/apps/gateway/Dockerfile index c88d2de..4ff401b 100644 --- a/apps/gateway/Dockerfile +++ b/apps/gateway/Dockerfile @@ -7,7 +7,12 @@ COPY go.mod go.sum ./ RUN go mod download COPY . . -RUN CGO_ENABLED=0 GOOS=linux go build -ldflags="-s -w" -o /gateway ./apps/gateway +# Release builds name themselves; logs and metrics carry these values. +ARG VERSION=dev +ARG COMMIT= +RUN CGO_ENABLED=0 GOOS=linux go build \ + -ldflags="-s -w -X github.com/dappnode/dappnode-nexus-gateway/pkg/buildinfo.Version=${VERSION} -X github.com/dappnode/dappnode-nexus-gateway/pkg/buildinfo.Commit=${COMMIT}" \ + -o /gateway ./apps/gateway # ---- Runtime stage ---- FROM alpine:3.20 diff --git a/apps/gateway/internal/adapters/http/handlers/chat_completions_handler.go b/apps/gateway/internal/adapters/http/handlers/chat_completions_handler.go index 2ad27f9..2644971 100644 --- a/apps/gateway/internal/adapters/http/handlers/chat_completions_handler.go +++ b/apps/gateway/internal/adapters/http/handlers/chat_completions_handler.go @@ -33,19 +33,19 @@ func NewChatCompletionsHandler(service *services.ChatCompletionsService, logger func (h *ChatCompletionsHandler) Handle(w http.ResponseWriter, r *http.Request) { token, err := ExtractBearerToken(r) if err != nil { - WriteErrorWithLog(w, r, h.logger, err) + rejectRequest(w, r, h.logger, domain.EndpointChatCompletions, err) return } body, err := io.ReadAll(io.LimitReader(r.Body, maxRequestBodySize)) if err != nil { - WriteErrorWithLog(w, r, h.logger, domain.ErrInvalidField("failed to read request body")) + rejectRequest(w, r, h.logger, domain.EndpointChatCompletions, domain.ErrInvalidField("failed to read request body")) return } genReq, err := mapper.ChatCompletionRequestToDomain(body) if err != nil { - WriteErrorWithLog(w, r, h.logger, err) + rejectRequest(w, r, h.logger, domain.EndpointChatCompletions, err) return } diff --git a/apps/gateway/internal/adapters/http/handlers/common.go b/apps/gateway/internal/adapters/http/handlers/common.go index f2a67e2..e14e619 100644 --- a/apps/gateway/internal/adapters/http/handlers/common.go +++ b/apps/gateway/internal/adapters/http/handlers/common.go @@ -7,6 +7,7 @@ import ( "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/http/dto" "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/http/middleware" + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/observability/metrics" "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/application/ports" "github.com/dappnode/dappnode-nexus-gateway/pkg/domain" ) @@ -43,29 +44,21 @@ func WriteErrorWithLog(w http.ResponseWriter, r *http.Request, logger ports.Logg "path", r.URL.Path, "error_type", fmt.Sprintf("%T", err), ) - gwErr = domain.ErrInternal("an internal error occurred") + gwErr = domain.ErrInternal("an internal error occurred").WithMeta("cause", domain.CauseUntypedError) } - // Log at warn for client errors, error for server errors. + // Log at warn for client errors, error for server errors. Provider errors + // can echo prompts: only the classification and log-safe metadata go in. + fields := []any{ + "request_id", middleware.GetRequestID(r.Context()), + "path", r.URL.Path, + "status", gwErr.HTTPStatus, + } + fields = append(fields, domain.DiagnosticFields(gwErr)...) if gwErr.HTTPStatus >= 500 { - fields := []any{ - "request_id", middleware.GetRequestID(r.Context()), - "status", gwErr.HTTPStatus, - "gateway_status", gwErr.HTTPStatus, - "error_code", gwErr.Code, - } - // Provider errors can echo prompts; only retain numeric upstream status. - if status, ok := gwErr.Metadata["upstream_status"].(int); ok { - fields = append(fields, "upstream_status", status) - } logger.Error("request failed", fields...) } else if gwErr.HTTPStatus >= 400 { - logger.Warn("request error", - "request_id", middleware.GetRequestID(r.Context()), - "status", gwErr.HTTPStatus, - "gateway_status", gwErr.HTTPStatus, - "error_code", gwErr.Code, - ) + logger.Warn("request error", fields...) } if gwErr.RetryAfter != "" { @@ -89,6 +82,14 @@ func errorDetail(err *domain.GatewayError) dto.ErrorDetail { return detail } +// rejectRequest answers a generation request that failed before reaching the +// generation service (missing key, unreadable or invalid body), counting it +// with the failures the service records itself. +func rejectRequest(w http.ResponseWriter, r *http.Request, logger ports.Logger, endpoint string, err error) { + metrics.RecordError(endpoint, metrics.UnresolvedModel, "unknown", metrics.PhaseRequest, err) + WriteErrorWithLog(w, r, logger, err) +} + // ExtractBearerToken extracts the bearer token from the Authorization header. func ExtractBearerToken(r *http.Request) (string, error) { auth := r.Header.Get("Authorization") diff --git a/apps/gateway/internal/adapters/http/handlers/confidential_chat_completions_handler.go b/apps/gateway/internal/adapters/http/handlers/confidential_chat_completions_handler.go index 90c9bf3..6e33641 100644 --- a/apps/gateway/internal/adapters/http/handlers/confidential_chat_completions_handler.go +++ b/apps/gateway/internal/adapters/http/handlers/confidential_chat_completions_handler.go @@ -80,20 +80,20 @@ func (h *ConfidentialChatCompletionsHandler) Handle(w http.ResponseWriter, r *ht w.Header().Set("Cache-Control", "no-store, no-transform") if r.Body == nil || r.Body == http.NoBody || r.ContentLength == 0 { - WriteErrorWithLog(w, r, h.logger, domain.ErrInvalidField("confidential request body is required")) + rejectRequest(w, r, h.logger, domain.EndpointChatCompletions, domain.ErrInvalidField("confidential request body is required")) return } encapsulatedKeys := r.Header.Values(protocol.EncapsulatedKeyHeader) if len(encapsulatedKeys) == 0 { - WriteErrorWithLog(w, r, h.logger, domain.ErrInvalidField("confidential endpoint requires an EHBP-encrypted body")) + rejectRequest(w, r, h.logger, domain.EndpointChatCompletions, domain.ErrInvalidField("confidential endpoint requires an EHBP-encrypted body")) return } if len(encapsulatedKeys) != 1 || !isCanonicalEHBPKey(encapsulatedKeys[0]) { - WriteErrorWithLog(w, r, h.logger, domain.ErrInvalidField("invalid EHBP encapsulated key header")) + rejectRequest(w, r, h.logger, domain.EndpointChatCompletions, domain.ErrInvalidField("invalid EHBP encapsulated key header")) return } if r.ContentLength > confidentialMaxEHBPBody { - WriteErrorWithLog(w, r, h.logger, domain.ErrInvalidField("encrypted request body exceeds the maximum size")) + rejectRequest(w, r, h.logger, domain.EndpointChatCompletions, domain.ErrInvalidField("encrypted request body exceeds the maximum size")) return } @@ -118,47 +118,48 @@ func isCanonicalEHBPKey(encoded string) bool { func (h *ConfidentialChatCompletionsHandler) handleDecrypted(w http.ResponseWriter, r *http.Request) { body, err := io.ReadAll(io.LimitReader(r.Body, confidentialEnvelopeLimit+1)) if err != nil { - WriteErrorWithLog(w, r, h.logger, domain.ErrInvalidField("failed to read confidential request")) + rejectRequest(w, r, h.logger, domain.EndpointChatCompletions, domain.ErrInvalidField("failed to read confidential request")) return } if len(body) > confidentialEnvelopeLimit { - WriteErrorWithLog(w, r, h.logger, domain.ErrInvalidField("confidential request body exceeds the maximum size")) + rejectRequest(w, r, h.logger, domain.EndpointChatCompletions, domain.ErrInvalidField("confidential request body exceeds the maximum size")) return } envelope, err := decodeConfidentialEnvelope(body) if err != nil { - WriteErrorWithLog(w, r, h.logger, domain.ErrInvalidField(err.Error())) + rejectRequest(w, r, h.logger, domain.EndpointChatCompletions, domain.ErrInvalidField(err.Error())) return } now := h.now().UTC() issuedAt, err := time.Parse(time.RFC3339, envelope.IssuedAt) if err != nil { - WriteErrorWithLog(w, r, h.logger, domain.ErrInvalidField("issued_at must be an RFC3339 timestamp")) + rejectRequest(w, r, h.logger, domain.EndpointChatCompletions, domain.ErrInvalidField("issued_at must be an RFC3339 timestamp")) return } if issuedAt.Before(now.Add(-confidentialMaxAge)) { - WriteErrorWithLog(w, r, h.logger, domain.ErrInvalidField("confidential request is stale")) + rejectRequest(w, r, h.logger, domain.EndpointChatCompletions, domain.ErrInvalidField("confidential request is stale")) return } if issuedAt.After(now.Add(confidentialFutureSkew)) { - WriteErrorWithLog(w, r, h.logger, domain.ErrInvalidField("confidential request timestamp is in the future")) + rejectRequest(w, r, h.logger, domain.EndpointChatCompletions, domain.ErrInvalidField("confidential request timestamp is in the future")) return } parsedID, err := uuid.Parse(envelope.RequestID) if err != nil || parsedID == uuid.Nil || parsedID.String() != envelope.RequestID { - WriteErrorWithLog(w, r, h.logger, domain.ErrInvalidField("request_id must be a canonical UUID")) + rejectRequest(w, r, h.logger, domain.EndpointChatCompletions, domain.ErrInvalidField("request_id must be a canonical UUID")) return } switch h.replays.checkAndStore(envelope.RequestID, now) { case replayDuplicate: - WriteErrorWithLog(w, r, h.logger, domain.ErrConflict("confidential request was already processed")) + rejectRequest(w, r, h.logger, domain.EndpointChatCompletions, domain.ErrConflict("confidential request was already processed")) return case replayCapacityReached: - WriteErrorWithLog(w, r, h.logger, domain.ErrProviderUnavailable("confidential replay protection")) + rejectRequest(w, r, h.logger, domain.EndpointChatCompletions, domain.ErrProviderUnavailable("confidential replay protection").WithMeta( + "fault", domain.FaultNexus, "dependency", "replay_store", "cause", domain.CauseReplayStoreFull)) return } diff --git a/apps/gateway/internal/adapters/http/handlers/reject_metrics_test.go b/apps/gateway/internal/adapters/http/handlers/reject_metrics_test.go new file mode 100644 index 0000000..98a674f --- /dev/null +++ b/apps/gateway/internal/adapters/http/handlers/reject_metrics_test.go @@ -0,0 +1,57 @@ +package handlers + +import ( + "net/http/httptest" + "strings" + "testing" + + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/observability/metrics" + "github.com/dappnode/dappnode-nexus-gateway/pkg/domain" + "github.com/prometheus/client_golang/prometheus/testutil" +) + +// Requests the handler rejects before the generation service sees them are +// counted with the service's own failures. +func TestChatHandlerCountsRejectedRequests(t *testing.T) { + logger := &privacyLogger{} + handler := NewChatCompletionsHandler(nil, logger) + for name, tc := range map[string]struct { + auth, body, code string + }{ + "missing key": {body: `{"model":"m","messages":[]}`, code: domain.ErrCodeInvalidAPIKey}, + "invalid body": {auth: "Bearer sk-test", body: `{`, code: domain.ErrCodeInvalidField}, + } { + t.Run(name, func(t *testing.T) { + counter := metrics.ErrorsTotal.WithLabelValues(domain.EndpointChatCompletions, metrics.UnresolvedModel, "unknown", + metrics.PhaseRequest, domain.FaultCaller, tc.code, "none") + before := testutil.ToFloat64(counter) + req := httptest.NewRequest("POST", "/v1/chat/completions", strings.NewReader(tc.body)) + if tc.auth != "" { + req.Header.Set("Authorization", tc.auth) + } + handler.Handle(httptest.NewRecorder(), req) + if got := testutil.ToFloat64(counter) - before; got != 1 { + t.Fatalf("errors_total delta = %v, want 1", got) + } + }) + } +} + +// Error lines carry the classification, and provider text that is not an +// identifier never reaches the log. +func TestErrorLogsCarryClassification(t *testing.T) { + const secret = "PRIVATE PROMPT CANARY" + logger := &privacyLogger{} + err := domain.ErrInternal("an internal error occurred").WithMeta("dependency", "pii", "cause", domain.CausePIIFilterTimeout, + "upstream_code", secret, "account_id", "acc_1") + WriteErrorWithLog(httptest.NewRecorder(), httptest.NewRequest("POST", "/v1/chat/completions", nil), logger, err) + out := logger.output.String() + for _, want := range []string{"fault nexus", "cause pii_filter_timeout", "dependency pii", "account_id acc_1"} { + if !strings.Contains(out, want) { + t.Errorf("log %q lacks %q", out, want) + } + } + if strings.Contains(out, secret) { + t.Fatal("error log leaked provider text") + } +} diff --git a/apps/gateway/internal/adapters/http/handlers/systemone_handler.go b/apps/gateway/internal/adapters/http/handlers/systemone_handler.go index 91792e4..8abc4e8 100644 --- a/apps/gateway/internal/adapters/http/handlers/systemone_handler.go +++ b/apps/gateway/internal/adapters/http/handlers/systemone_handler.go @@ -22,17 +22,17 @@ func NewSystemOneHandler(service *services.GenerateService, logger ports.Logger) func (h *SystemOneHandler) Handle(w http.ResponseWriter, r *http.Request) { token, err := ExtractBearerToken(r) if err != nil { - WriteErrorWithLog(w, r, h.logger, err) + rejectRequest(w, r, h.logger, domain.EndpointSystemOne, err) return } body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxRequestBodySize)) if err != nil { - WriteErrorWithLog(w, r, h.logger, domain.ErrInvalidField("request body unreadable or too large")) + rejectRequest(w, r, h.logger, domain.EndpointSystemOne, domain.ErrInvalidField("request body unreadable or too large")) return } req, err := mapper.SystemOneRequestToDomain(body) if err != nil { - WriteErrorWithLog(w, r, h.logger, err) + rejectRequest(w, r, h.logger, domain.EndpointSystemOne, err) return } call, err := h.service.Proxy(r.Context(), domain.EndpointSystemOne, body, req, token) diff --git a/apps/gateway/internal/adapters/http/middleware/metrics.go b/apps/gateway/internal/adapters/http/middleware/metrics.go index d1a4e72..de71bca 100644 --- a/apps/gateway/internal/adapters/http/middleware/metrics.go +++ b/apps/gateway/internal/adapters/http/middleware/metrics.go @@ -12,11 +12,18 @@ import ( // intentionally placed as the outermost middleware so the recorded duration // covers the full handler chain. // -// The route label is read from r.Pattern after the handler runs, because the -// stdlib ServeMux only populates it when dispatching to a matched route -// handler. Requests that match no route use the fixed "unmatched" label so a -// flood of 404s cannot create unbounded series. -func Metrics(next http.Handler) http.Handler { +// The route label comes from route, which names the mux pattern a request +// matches: inner middleware passes the mux a copy of the request, so +// r.Pattern is never set on the request seen here. Requests that match no +// route use the fixed "unmatched" label so a flood of 404s cannot create +// unbounded series. +func Metrics(route func(*http.Request) string) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return metricsHandler(next, route) + } +} + +func metricsHandler(next http.Handler, routeOf func(*http.Request) string) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { mw := &metricsWriter{ResponseWriter: w, status: http.StatusOK} @@ -26,7 +33,7 @@ func Metrics(next http.Handler) http.Handler { duration := time.Since(start).Seconds() metrics.HTTPInflight.Dec() - route := r.Pattern + route := routeOf(r) if route == "" { route = "unmatched" } diff --git a/apps/gateway/internal/adapters/http/router.go b/apps/gateway/internal/adapters/http/router.go index 29bb22e..701fee9 100644 --- a/apps/gateway/internal/adapters/http/router.go +++ b/apps/gateway/internal/adapters/http/router.go @@ -42,7 +42,12 @@ func NewRouter( handler = middleware.Logging(logger)(handler) handler = middleware.Recovery(logger)(handler) handler = middleware.RequestID(handler) - handler = middleware.Metrics(handler) + // Middleware hands the mux a copy of the request, so the route pattern is + // read from the mux rather than from the request the metrics see. + handler = middleware.Metrics(func(r *http.Request) string { + _, pattern := mux.Handler(r) + return pattern + })(handler) return handler } diff --git a/apps/gateway/internal/adapters/http/router_test.go b/apps/gateway/internal/adapters/http/router_test.go index ff37283..9c95f10 100644 --- a/apps/gateway/internal/adapters/http/router_test.go +++ b/apps/gateway/internal/adapters/http/router_test.go @@ -7,6 +7,8 @@ import ( "testing" "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/http/handlers" + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/observability/metrics" + "github.com/prometheus/client_golang/prometheus/testutil" ) type routerTestLogger struct{} @@ -56,3 +58,23 @@ func TestNewRouter_PreservesLegacyChatRouteWithoutConfidentialHandler(t *testing t.Fatalf("unauthenticated EHBP key discovery status = %d, want 404", keyDiscoveryResponse.Code) } } + +// Requests are counted under the route they matched, though the middleware +// in between hands the mux a copy of the request. +func TestNewRouter_LabelsMetricsWithTheMatchedRoute(t *testing.T) { + logger := &routerTestLogger{} + router := NewRouter(handlers.NewHealthHandler(), handlers.NewModelsHandler(nil), handlers.NewChatCompletionsHandler(nil, logger), nil, nil, nil, nil, logger) + health := metrics.HTTPRequestsTotal.WithLabelValues("GET /healthz", http.MethodGet, "2xx") + unmatched := metrics.HTTPRequestsTotal.WithLabelValues("unmatched", http.MethodGet, "4xx") + healthBefore, unmatchedBefore := testutil.ToFloat64(health), testutil.ToFloat64(unmatched) + + router.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/healthz", nil)) + router.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/nope", nil)) + + if got := testutil.ToFloat64(health) - healthBefore; got != 1 { + t.Fatalf("GET /healthz count delta = %v, want 1", got) + } + if got := testutil.ToFloat64(unmatched) - unmatchedBefore; got != 1 { + t.Fatalf("unmatched count delta = %v, want 1", got) + } +} diff --git a/apps/gateway/internal/adapters/metering/client.go b/apps/gateway/internal/adapters/metering/client.go index 00155f8..283a270 100644 --- a/apps/gateway/internal/adapters/metering/client.go +++ b/apps/gateway/internal/adapters/metering/client.go @@ -65,7 +65,7 @@ func (c *Client) Reserve(ctx context.Context, auth domain.AuthContext, endpoint return "", err } if res.ReservationID == "" { - return "", meteringInternalError("reason", "missing reservation_id") + return "", meteringInternalError("cause", domain.CauseMeteringBadResponse, "reason", "missing reservation_id") } if res.LeaseMS > 0 { c.reservations.Store(res.ReservationID, newLease(sent, res.LeaseMS)) @@ -194,7 +194,7 @@ func (c *Client) requestJSONOnce(ctx context.Context, method, path string, paylo if ctx.Err() != nil { return ctx.Err() } - return newRetryableRequestError(meteringInternalError("reason", err.Error())) + return newRetryableRequestError(meteringInternalError("cause", domain.CauseMeteringUnreachable, "reason", err.Error())) } defer res.Body.Close() if res.StatusCode < 200 || res.StatusCode >= 300 { @@ -215,14 +215,14 @@ func (c *Client) requestJSONOnce(ctx context.Context, method, path string, paylo metrics.MeteringRequests.WithLabelValues(op, "protocol_error").Inc() metrics.MeteringDuration.WithLabelValues(op).Observe(time.Since(start).Seconds()) return newRetryableRequestError( - meteringInternalError("reason", "missing response body"), + meteringInternalError("cause", domain.CauseMeteringBadResponse, "reason", "missing response body"), ) } if err := decodeJSONResponse(res.Body, out); err != nil { metrics.MeteringRequests.WithLabelValues(op, "protocol_error").Inc() metrics.MeteringDuration.WithLabelValues(op).Observe(time.Since(start).Seconds()) return newRetryableRequestError( - meteringInternalError("reason", "invalid response body: "+err.Error()), + meteringInternalError("cause", domain.CauseMeteringBadResponse, "reason", "invalid response body: "+err.Error()), ) } } @@ -310,7 +310,7 @@ func userFacingMeteringError(method, path string, status int, envelope errorResp } func meteringResponseError(status int, code string) *domain.GatewayError { - err := meteringInternalError("metering_status", status) + err := meteringInternalError("cause", domain.CauseMeteringErrorStatus, "metering_status", status) if code != "" { err = err.WithMeta("metering_code", code) } diff --git a/apps/gateway/internal/adapters/metering/client_test.go b/apps/gateway/internal/adapters/metering/client_test.go index 0510f48..cabf6ca 100644 --- a/apps/gateway/internal/adapters/metering/client_test.go +++ b/apps/gateway/internal/adapters/metering/client_test.go @@ -62,6 +62,7 @@ func TestClientReserve_ExhaustedServerErrorsReturnInternalError(t *testing.T) { client := NewClient(server.URL, "token", time.Second) _, err := client.Reserve(context.Background(), testAuth(), domain.EndpointChatCompletions, domain.GenerateRequest{PublicModelID: "model"}, testModel(), "gw-1") assertInternalMeteringError(t, err) + assertMeteringCause(t, err, domain.CauseMeteringErrorStatus) if got := attempts.Load(); got != 3 { t.Fatalf("attempts = %d, want 3", got) } @@ -75,6 +76,17 @@ func TestClientReserve_TransportFailureReturnsInternalError(t *testing.T) { client := NewClient(serverURL, "token", time.Second) _, err := client.Reserve(context.Background(), testAuth(), domain.EndpointChatCompletions, domain.GenerateRequest{PublicModelID: "model"}, testModel(), "gw-1") assertInternalMeteringError(t, err) + assertMeteringCause(t, err, domain.CauseMeteringUnreachable) +} + +// assertMeteringCause checks that a metering failure names its cause, so an +// internal error can be traced to metering without reading free text. +func assertMeteringCause(t *testing.T, err error, want string) { + t.Helper() + var gwErr *domain.GatewayError + if !errors.As(err, &gwErr) || gwErr.Metadata["cause"] != want || gwErr.Metadata["dependency"] != "metering" { + t.Fatalf("error = %v, metadata = %v; want dependency metering, cause %s", err, gwErr.Metadata, want) + } } func TestClientReserve_RetriesAmbiguousResponseWithStableRequestID(t *testing.T) { diff --git a/apps/gateway/internal/adapters/metering/runtime_client.go b/apps/gateway/internal/adapters/metering/runtime_client.go index 3fd359b..ba39c2d 100644 --- a/apps/gateway/internal/adapters/metering/runtime_client.go +++ b/apps/gateway/internal/adapters/metering/runtime_client.go @@ -311,7 +311,7 @@ func validateRuntimeProof(proof runtimeProof, expectedAccountID, expectedProvide } func runtimeContractError(contract, reason string) *domain.GatewayError { - return meteringInternalError("contract", contract, "reason", reason) + return meteringInternalError("cause", domain.CauseMeteringBadResponse, "contract", contract, "reason", reason) } func jsonObject(raw json.RawMessage) bool { diff --git a/apps/gateway/internal/adapters/observability/metrics/prometheus.go b/apps/gateway/internal/adapters/observability/metrics/prometheus.go index 817950b..370a4f8 100644 --- a/apps/gateway/internal/adapters/observability/metrics/prometheus.go +++ b/apps/gateway/internal/adapters/observability/metrics/prometheus.go @@ -4,8 +4,11 @@ package metrics import ( + "errors" "net/http" + "strconv" + "github.com/dappnode/dappnode-nexus-gateway/pkg/buildinfo" "github.com/dappnode/dappnode-nexus-gateway/pkg/domain" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promhttp" @@ -187,6 +190,69 @@ var ( ) ) +// Error and build metrics. ErrorsTotal classifies every failed generation, so +// an error rate can count provider and Nexus faults without caller mistakes. +var ( + ErrorsTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "nexus_gateway_errors_total", + Help: "Total failed generations, by endpoint, model, provider, phase (request|provider|stream), fault (caller|provider|nexus|unclear), error code and upstream status.", + }, + []string{"endpoint", "model", "provider", "phase", "fault", "error_code", "upstream_status"}, + ) + + FallbacksTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "nexus_gateway_fallbacks_total", + Help: "Total switches to a model's fallback provider, by endpoint, model, primary provider, fallback provider and the primary's fault and error code.", + }, + []string{"endpoint", "model", "provider", "fallback_provider", "fault", "error_code"}, + ) + + BuildInfo = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "nexus_gateway_build_info", + Help: "Always 1, labeled by the running gateway's version and commit.", + }, + []string{"version", "commit"}, + ) +) + +// Error phases: where in a request the failure happened. +const ( + PhaseRequest = "request" // Before any provider call: auth, routing, validation, PII, credit. + PhaseProvider = "provider" // The provider failed before any output reached the client. + PhaseStream = "stream" // The response failed after output began. +) + +// UnresolvedModel labels requests whose model was never resolved, so caller +// supplied names cannot create metric series. +const UnresolvedModel = "unresolved" + +// RecordError counts a failed generation. +func RecordError(endpoint, model, provider, phase string, err error) { + upstream := "none" + if status := domain.UpstreamStatus(err); status > 0 { + upstream = strconv.Itoa(status) + } + ErrorsTotal.WithLabelValues(endpoint, model, provider, phase, domain.FaultOf(err), errorCode(err), upstream).Inc() +} + +// RecordFallback counts a primary provider failure that the fallback +// provider took over, which the error metric never sees when the fallback +// succeeds. +func RecordFallback(endpoint, model, provider, fallbackProvider string, err error) { + FallbacksTotal.WithLabelValues(endpoint, model, provider, fallbackProvider, domain.FaultOf(err), errorCode(err)).Inc() +} + +func errorCode(err error) string { + var gwErr *domain.GatewayError + if errors.As(err, &gwErr) { + return gwErr.Code + } + return domain.ErrCodeInternalError +} + func init() { prometheus.MustRegister( HTTPRequestsTotal, @@ -207,7 +273,11 @@ func init() { AttestationOperationsTotal, AttestationDuration, ProviderRetries, + ErrorsTotal, + FallbacksTotal, + BuildInfo, ) + BuildInfo.WithLabelValues(buildinfo.Version, buildinfo.Commit).Set(1) } // Handler returns the Prometheus metrics HTTP handler. diff --git a/apps/gateway/internal/adapters/providers/openai/adapter.go b/apps/gateway/internal/adapters/providers/openai/adapter.go index 1f6710a..c1b6862 100644 --- a/apps/gateway/internal/adapters/providers/openai/adapter.go +++ b/apps/gateway/internal/adapters/providers/openai/adapter.go @@ -39,6 +39,7 @@ func NewAdapter(timeout time.Duration, logger ...ports.Logger) *Adapter { func missingProviderCredentialError(providerName string) *domain.GatewayError { return domain.ErrInternal("an internal error occurred").WithMeta( "provider", providerName, + "cause", domain.CauseProviderKeyMissing, "reason", "provider API key is not configured", ) } diff --git a/apps/gateway/internal/adapters/providers/tinfoil/adapter.go b/apps/gateway/internal/adapters/providers/tinfoil/adapter.go index 13f7d22..9238017 100644 --- a/apps/gateway/internal/adapters/providers/tinfoil/adapter.go +++ b/apps/gateway/internal/adapters/providers/tinfoil/adapter.go @@ -65,6 +65,7 @@ func NewAdapterWithFactory(timeout time.Duration, factory VerifiedClientFactory, func missingProviderCredentialError() *domain.GatewayError { return domain.ErrInternal("an internal error occurred").WithMeta( "provider", providerName, + "cause", domain.CauseProviderKeyMissing, "reason", "provider API key is not configured", ) } diff --git a/apps/gateway/internal/adapters/providers/typesafe/adapter.go b/apps/gateway/internal/adapters/providers/typesafe/adapter.go index c6ca9ef..8320126 100644 --- a/apps/gateway/internal/adapters/providers/typesafe/adapter.go +++ b/apps/gateway/internal/adapters/providers/typesafe/adapter.go @@ -20,7 +20,11 @@ func NewAdapter(timeout time.Duration) *Adapter { return &Adapter{client: openai func (a *Adapter) Complete(ctx context.Context, raw []byte, model domain.PublicModel) ([]byte, *domain.TinfoilTransportProof, error) { apiKey := os.Getenv(model.ProviderConfig.APIKeySecretRef) if apiKey == "" { - return nil, nil, domain.ErrInternal("provider API key is not configured") + return nil, nil, domain.ErrInternal("an internal error occurred").WithMeta( + "provider", "typesafe", + "cause", domain.CauseProviderKeyMissing, + "reason", "provider API key is not configured", + ) } var body map[string]any dec := json.NewDecoder(bytes.NewReader(raw)) diff --git a/apps/gateway/internal/adapters/router/http_client.go b/apps/gateway/internal/adapters/router/http_client.go index e878b70..81db452 100644 --- a/apps/gateway/internal/adapters/router/http_client.go +++ b/apps/gateway/internal/adapters/router/http_client.go @@ -77,7 +77,7 @@ func (c *Client) Route(ctx context.Context, req domain.RouteRequest) (domain.Rou if err != nil { metrics.RouterRequests.WithLabelValues("transport_error").Inc() metrics.RouterDuration.Observe(time.Since(start).Seconds()) - return domain.RouteDecision{}, routerInternalError("reason", err.Error()) + return domain.RouteDecision{}, routerInternalError("cause", domain.CauseRouterUnreachable, "reason", err.Error()) } defer resp.Body.Close() @@ -88,14 +88,14 @@ func (c *Client) Route(ctx context.Context, req domain.RouteRequest) (domain.Rou if decodeErr := json.NewDecoder(resp.Body).Decode(&errResp); decodeErr == nil && errResp.Error.Code != "" { return domain.RouteDecision{}, mapRouterError(req.RouterID, resp.StatusCode, errResp) } - return domain.RouteDecision{}, routerInternalError("router_status", resp.StatusCode) + return domain.RouteDecision{}, routerInternalError("cause", domain.CauseRouterErrorStatus, "router_status", resp.StatusCode) } var out routeResponse if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { metrics.RouterRequests.WithLabelValues("protocol_error").Inc() metrics.RouterDuration.Observe(time.Since(start).Seconds()) - return domain.RouteDecision{}, routerInternalError("reason", err.Error()) + return domain.RouteDecision{}, routerInternalError("cause", domain.CauseRouterBadResponse, "reason", err.Error()) } metrics.RouterRequests.WithLabelValues("success").Inc() @@ -124,6 +124,7 @@ func mapRouterError(routerID string, status int, errResp errorResponse) error { return domain.ErrInvalidField(errResp.Error.Message) default: return routerInternalError( + "cause", domain.CauseRouterErrorStatus, "router_status", status, "router_code", errResp.Error.Code, "reason", errResp.Error.Message, diff --git a/apps/gateway/internal/application/services/error_classification_test.go b/apps/gateway/internal/application/services/error_classification_test.go new file mode 100644 index 0000000..f4e241f --- /dev/null +++ b/apps/gateway/internal/application/services/error_classification_test.go @@ -0,0 +1,76 @@ +package services + +import ( + "context" + "testing" + + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/observability/metrics" + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/application/ports" + "github.com/dappnode/dappnode-nexus-gateway/pkg/domain" + "github.com/prometheus/client_golang/prometheus/testutil" +) + +// A primary provider failure that the fallback covers is counted, though the +// request succeeds and the error metric never sees it. +func TestProxy_CountsFallbacks(t *testing.T) { + primary := &stubProvider{err: domain.ErrProviderUnavailable("primary")} + svc := newFallbackGenerateService(&stubUsageMeter{}, primary, &stubProvider{}) + counter := metrics.FallbacksTotal.WithLabelValues(domain.EndpointChatCompletions, "model", "primary", "fallback", + domain.FaultProvider, domain.ErrCodeProviderUnavailable) + before := testutil.ToFloat64(counter) + + if _, err := svc.Proxy(context.Background(), domain.EndpointChatCompletions, []byte(`{}`), domain.GenerateRequest{PublicModelID: "model"}, "sk-test"); err != nil { + t.Fatal(err) + } + if got := testutil.ToFloat64(counter) - before; got != 1 { + t.Fatalf("fallbacks_total delta = %v, want 1", got) + } +} + +// A provider name this build doesn't know is Nexus's configuration, not the +// provider's outage, though the client still sees provider_unavailable. +func TestProxy_UnregisteredProviderIsNexusFault(t *testing.T) { + svc := proxyService(&resultMeter{}, "", map[string]ports.Provider{}, false) + counter := metrics.ErrorsTotal.WithLabelValues(domain.EndpointChatCompletions, "zai-org/glm-5.3-flash", "primary", + metrics.PhaseProvider, domain.FaultNexus, domain.ErrCodeProviderUnavailable, "none") + before := testutil.ToFloat64(counter) + + _, err := svc.Proxy(context.Background(), domain.EndpointChatCompletions, []byte(`{}`), domain.GenerateRequest{PublicModelID: "zai-org/glm-5.3-flash"}, "key") + gwErr, ok := err.(*domain.GatewayError) + if !ok || gwErr.Code != domain.ErrCodeProviderUnavailable || gwErr.Metadata["cause"] != domain.CauseProviderNotRegistered { + t.Fatalf("error = %v", err) + } + if got := testutil.ToFloat64(counter) - before; got != 1 { + t.Fatalf("errors_total delta = %v, want 1", got) + } +} + +// When the client leaves during PII masking, the canceled analyzer call is +// the client's cancellation, not a Nexus failure to fix. +func TestProxy_ClientLeavingDuringPIIIsNotNexusFault(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + svc, _ := piiService(&fakePIIFilter{enabled: true, err: context.Canceled}, domain.APIKeyPIIModeHigh, &stubProvider{}) + + _, err := svc.Proxy(ctx, domain.EndpointChatCompletions, []byte(`{"messages":[{"role":"user","content":"hi"}]}`), domain.GenerateRequest{PublicModelID: "zai-org/glm-5.3-flash"}, "key") + gwErr, ok := err.(*domain.GatewayError) + if !ok || gwErr.Code != domain.ErrCodeClientCanceled || domain.FaultOf(err) != domain.FaultCaller { + t.Fatalf("error = %v, want client_canceled", err) + } +} + +func TestClientGoneKeepsCallerAndProviderErrors(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + for _, err := range []error{domain.ErrUnsupportedModel("m"), domain.ErrInsufficientBalance(), domain.ErrProviderTimeout("p")} { + if got := clientGone(ctx, err); got != err { + t.Errorf("clientGone(%v) = %v, want unchanged", err, got) + } + } + if got := clientGone(ctx, domain.ErrInternal("x")); domain.FaultOf(got) != domain.FaultCaller { + t.Errorf("internal error after the client left = %v, want client_canceled", got) + } + if got := clientGone(context.Background(), domain.ErrInternal("x")); domain.FaultOf(got) != domain.FaultNexus { + t.Errorf("internal error with the client present = %v, want unchanged", got) + } +} diff --git a/apps/gateway/internal/application/services/generate_service.go b/apps/gateway/internal/application/services/generate_service.go index 14aea0e..feeac0f 100644 --- a/apps/gateway/internal/application/services/generate_service.go +++ b/apps/gateway/internal/application/services/generate_service.go @@ -96,14 +96,16 @@ func (s *GenerateService) logFallback(requestID string, primary, fallback domain if s.logger == nil { return } - s.logger.Warn("provider failed; trying fallback", + fields := []any{ "request_id", requestID, + "model", primary.PublicModelID, "provider", primary.ProviderConfig.ProviderName, "provider_model", primary.UpstreamModelName, "fallback_provider", fallback.ProviderConfig.ProviderName, "fallback_provider_model", fallback.UpstreamModelName, "error_type", fmt.Sprintf("%T", err), - ) + } + s.logger.Warn("provider failed; trying fallback", append(fields, domain.DiagnosticFields(err)...)...) } func (s *GenerateService) resolveModel(ctx context.Context, req domain.GenerateRequest) (domain.PublicModel, domain.GenerateRequest, error) { @@ -130,6 +132,7 @@ func (s *GenerateService) resolveModel(ctx context.Context, req domain.GenerateR if s.router == nil { return domain.PublicModel{}, req, domain.ErrInternal("an internal error occurred").WithMeta( "dependency", "router", + "cause", domain.CauseRouterMissing, "reason", "router client is not configured", ) } @@ -144,6 +147,7 @@ func (s *GenerateService) resolveModel(ctx context.Context, req domain.GenerateR if decision.PublicModelID == "" { return domain.PublicModel{}, req, domain.ErrInternal("an internal error occurred").WithMeta( "dependency", "router", + "cause", domain.CauseRouterEmptyModel, "reason", "router returned an empty public_model_id", ) } @@ -245,22 +249,49 @@ func (s *GenerateService) recordGeneration(outcome, endpoint string, req domain. // exists; otherwise it records the request as unrouted so no terminal outcome // is silently dropped. func (s *GenerateService) recordTerminalOutcome(outcome, endpoint string, req domain.GenerateRequest, model *domain.PublicModel, latencyMs int64) { - providerName := "unknown" - if model != nil { - providerName = model.ProviderConfig.ProviderName - } + modelLabel, providerName := metricLabels(req, model) stream := strconv.FormatBool(req.Stream) metrics.GenerationsTotal.WithLabelValues( endpoint, - req.PublicModelID, + modelLabel, providerName, stream, outcome, ).Inc() - metrics.GenerationDuration.WithLabelValues(endpoint, req.PublicModelID, providerName, stream). + metrics.GenerationDuration.WithLabelValues(endpoint, modelLabel, providerName, stream). Observe(float64(latencyMs) / 1000.0) } +// recordError classifies a failed generation in the error metric. +func (s *GenerateService) recordError(endpoint string, req domain.GenerateRequest, model *domain.PublicModel, phase string, err error) { + modelLabel, providerName := metricLabels(req, model) + metrics.RecordError(endpoint, modelLabel, providerName, phase, err) +} + +// metricLabels names the model and provider for metrics, by their catalog +// names. A model that never resolved is labeled as such: its name came from +// the caller. +func metricLabels(req domain.GenerateRequest, model *domain.PublicModel) (string, string) { + if model == nil { + return metrics.UnresolvedModel, "unknown" + } + if model.PublicModelID == "" { + return req.PublicModelID, model.ProviderConfig.ProviderName + } + return model.PublicModelID, model.ProviderConfig.ProviderName +} + +// clientGone reports a Nexus-side failure that happened after the client +// left as the client's cancellation: the canceled dependency call is a +// consequence, not a fault to fix. Caller and provider errors keep their +// meaning. +func clientGone(ctx context.Context, err error) error { + if err != nil && ctx.Err() != nil && domain.FaultOf(err) == domain.FaultNexus { + return domain.ErrClientCanceled() + } + return err +} + // buildErrorLogFields builds a structured log field slice for error conditions, // including request context, provider details, and numeric upstream status, never error messages or arbitrary metadata. func (s *GenerateService) buildErrorLogFields(ctx context.Context, requestID string, authCtx *domain.AuthContext, endpoint, publicModelID string, model domain.PublicModel, err error, latencyMs int64) []any { @@ -276,17 +307,9 @@ func (s *GenerateService) buildErrorLogFields(ctx context.Context, requestID str if authCtx != nil { fields = append(fields, "account_id", authCtx.Account.ID) } - // Merge structured upstream metadata from GatewayError if present. - var gwErr *domain.GatewayError - if errors.As(err, &gwErr) { - fields = append(fields, "error_code", gwErr.Code) - fields = append(fields, "gateway_status", gwErr.HTTPStatus) - // Upstream messages and arbitrary metadata may echo request contents. - if status, ok := gwErr.Metadata["upstream_status"].(int); ok { - fields = append(fields, "upstream_status", status) - } - } - return fields + // Classification and log-safe metadata only: upstream messages and + // arbitrary metadata may echo request contents. + return append(fields, domain.DiagnosticFields(err)...) } func (s *GenerateService) storeTinfoilProof(ctx context.Context, auth domain.AuthContext, model domain.PublicModel, result domain.GenerateResult) { diff --git a/apps/gateway/internal/application/services/generate_service_test.go b/apps/gateway/internal/application/services/generate_service_test.go index 35de009..b41aa8a 100644 --- a/apps/gateway/internal/application/services/generate_service_test.go +++ b/apps/gateway/internal/application/services/generate_service_test.go @@ -8,9 +8,11 @@ import ( "testing" "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/http/middleware" + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/observability/metrics" "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/application/ports" "github.com/dappnode/dappnode-nexus-gateway/pkg/domain" "github.com/google/uuid" + "github.com/prometheus/client_golang/prometheus/testutil" "github.com/shopspring/decimal" ) @@ -583,3 +585,44 @@ func TestRequestRejectionDoesNotChangeProvider(t *testing.T) { } } } + +// A request for a model that doesn't exist is the caller's mistake: it is +// classified as such, named in the log, and cannot create metric series. +func TestProxy_ClassifiesUnknownModelAsCallerFault(t *testing.T) { + svc := NewGenerateService( + &stubAuthService{authCtx: domain.AuthContext{ + Account: domain.Account{ID: "acc1", Status: domain.AccountStatusActive}, + APIKey: domain.APIKey{ID: "key1", Active: true}, + }}, + &stubModelCatalog{models: map[string]domain.PublicModel{}}, + nil, + nil, + &stubUsageMeter{}, + nil, + stubLogger{}, + ) + counter := metrics.ErrorsTotal.WithLabelValues(domain.EndpointChatCompletions, metrics.UnresolvedModel, "unknown", + metrics.PhaseRequest, domain.FaultCaller, domain.ErrCodeUnsupportedModel, "none") + before := testutil.ToFloat64(counter) + + _, err := svc.Proxy(context.Background(), domain.EndpointChatCompletions, []byte(`{}`), domain.GenerateRequest{ + PublicModelID: "deepseek/deepseek-v4-flash", + }, "sk-test") + + gwErr, ok := err.(*domain.GatewayError) + if !ok || gwErr.Code != domain.ErrCodeUnsupportedModel { + t.Fatalf("error = %v, want unsupported_model", err) + } + if gwErr.Metadata["requested_model"] != "deepseek/deepseek-v4-flash" || gwErr.Metadata["account_id"] != "acc1" { + t.Fatalf("metadata = %v", gwErr.Metadata) + } + if got := testutil.ToFloat64(counter) - before; got != 1 { + t.Fatalf("errors_total delta = %v, want 1", got) + } + if n := testutil.CollectAndCount(metrics.GenerationsTotal, "nexus_gateway_generations_total"); n == 0 { + t.Fatal("no generation series recorded") + } + if testutil.ToFloat64(metrics.GenerationsTotal.WithLabelValues(domain.EndpointChatCompletions, "deepseek/deepseek-v4-flash", "unknown", "false", metrics.OutcomeError)) != 0 { + t.Fatal("the caller's model name became a metric label") + } +} diff --git a/apps/gateway/internal/application/services/pii.go b/apps/gateway/internal/application/services/pii.go index 48069dd..155614e 100644 --- a/apps/gateway/internal/application/services/pii.go +++ b/apps/gateway/internal/application/services/pii.go @@ -6,10 +6,12 @@ import ( "encoding/json" "errors" "io" + "net" "sort" "strconv" "strings" + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/http/middleware" "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/application/ports" "github.com/dappnode/dappnode-nexus-gateway/pkg/domain" ) @@ -44,11 +46,15 @@ func (s *GenerateService) maskBody(ctx context.Context, raw []byte, piiMode, end surfaceCounts: make(map[string]int), } if err := masker.maskBody(body, endpoint); err != nil { - s.logger.Warn("pii filter error", "error", err, "fail_open", s.piiFailOpen) + if ctx.Err() != nil { + return nil, nil, domain.ErrClientCanceled() // The client left; the analyzer did not fail. + } + cause := piiFailureCause(err) + s.logger.Warn("pii filter error", "request_id", middleware.GetRequestID(ctx), "error", err, "cause", cause, "fail_open", s.piiFailOpen) if s.piiFailOpen { return raw, nil, nil } - return nil, nil, domain.ErrInternal("an internal error occurred") + return nil, nil, domain.ErrInternal("an internal error occurred").WithMeta("dependency", "pii", "cause", cause) } if masker.mapping.Len() == 0 { return raw, nil, nil @@ -61,11 +67,20 @@ func (s *GenerateService) maskBody(ctx context.Context, raw []byte, piiMode, end ) masked, err := marshalNoEscape(body) if err != nil { - return nil, nil, domain.ErrInternal("an internal error occurred") + return nil, nil, domain.ErrInternal("an internal error occurred").WithMeta("dependency", "pii", "cause", domain.CausePIIFilterFailed) } return masked, masker.mapping, nil } +// piiFailureCause tells a PII analyzer timeout from other analyzer failures. +func piiFailureCause(err error) string { + var netErr net.Error + if errors.Is(err, context.DeadlineExceeded) || (errors.As(err, &netErr) && netErr.Timeout()) { + return domain.CausePIIFilterTimeout + } + return domain.CausePIIFilterFailed +} + func (m *requestPIIMasker) maskBody(body map[string]any, endpoint string) error { if endpoint == domain.EndpointSystemOne { masked, _, err := m.maskAnyWithChanged("state", body["state"]) diff --git a/apps/gateway/internal/application/services/pii_test.go b/apps/gateway/internal/application/services/pii_test.go index eb5e846..66a8286 100644 --- a/apps/gateway/internal/application/services/pii_test.go +++ b/apps/gateway/internal/application/services/pii_test.go @@ -3,6 +3,7 @@ package services import ( "context" "encoding/json" + "fmt" "io" "strings" "testing" @@ -306,3 +307,27 @@ func TestSanitizeErrorWithPIIMapping_MasksKnownOriginalsInMessageAndMetadata(t * t.Fatalf("not sanitized: %+v", sanitized) } } + +// A failing PII analyzer is Nexus's fault, and the error says whether it +// timed out, so the report can tell a slow analyzer from a broken one. +func TestPII_FailureNamesItsCause(t *testing.T) { + for name, tc := range map[string]struct { + err error + want string + }{ + "timeout": {err: fmt.Errorf("presidio: call analyzer: %w", context.DeadlineExceeded), want: domain.CausePIIFilterTimeout}, + "failure": {err: io.ErrUnexpectedEOF, want: domain.CausePIIFilterFailed}, + } { + t.Run(name, func(t *testing.T) { + svc, _ := piiService(&fakePIIFilter{enabled: true, err: tc.err}, domain.APIKeyPIIModeHigh, &stubProvider{}) + _, err := svc.Proxy(context.Background(), domain.EndpointChatCompletions, []byte(`{"messages":[{"role":"user","content":"hi"}]}`), domain.GenerateRequest{PublicModelID: "zai-org/glm-5.3-flash"}, "key") + gwErr, ok := err.(*domain.GatewayError) + if !ok || gwErr.Code != domain.ErrCodeInternalError { + t.Fatalf("error = %v", err) + } + if gwErr.Metadata["cause"] != tc.want || gwErr.Metadata["dependency"] != "pii" || domain.FaultOf(err) != domain.FaultNexus { + t.Fatalf("metadata = %v", gwErr.Metadata) + } + }) + } +} diff --git a/apps/gateway/internal/application/services/proxy.go b/apps/gateway/internal/application/services/proxy.go index 36201d0..5dde257 100644 --- a/apps/gateway/internal/application/services/proxy.go +++ b/apps/gateway/internal/application/services/proxy.go @@ -43,9 +43,19 @@ func (s *GenerateService) Proxy(ctx context.Context, endpoint string, raw []byte authCtx, err := s.auth.AuthenticateAPIKey(ctx, bearerToken) if err != nil { + err = clientGone(ctx, err) s.recordTerminalOutcome(metrics.OutcomeError, endpoint, req, nil, time.Since(start).Milliseconds()) + s.recordError(endpoint, req, nil, metrics.PhaseRequest, err) return nil, err } + // reject names the account and model on an error that ends the request + // before any provider call, for the handler's log line. + reject := func(err error, model *domain.PublicModel) error { + if model == nil { + return domain.Annotate(err, "account_id", authCtx.Account.ID, "requested_model", req.PublicModelID) + } + return domain.Annotate(err, "account_id", authCtx.Account.ID, "model", model.PublicModelID, "provider", model.ProviderConfig.ProviderName) + } var model domain.PublicModel execReq := req if endpoint == domain.EndpointSystemOne { @@ -54,29 +64,38 @@ func (s *GenerateService) Proxy(ctx context.Context, endpoint string, raw []byte model, execReq, err = s.resolveModel(ctx, req) } if err != nil { + err = clientGone(ctx, err) s.recordTerminalOutcome(metrics.OutcomeError, endpoint, req, nil, time.Since(start).Milliseconds()) - return nil, err + s.recordError(endpoint, req, nil, metrics.PhaseRequest, err) + return nil, reject(err, nil) } if err := s.validateRequest(endpoint, execReq, model); err != nil { s.recordTerminalOutcome(metrics.OutcomeError, endpoint, execReq, &model, time.Since(start).Milliseconds()) + s.recordError(endpoint, execReq, &model, metrics.PhaseRequest, err) s.recordFailure(ctx, nil, &authCtx, endpoint, &execReq, &model, err, nil, time.Since(start).Milliseconds()) - return nil, err + return nil, reject(err, &model) } masked, mapping, err := s.maskBody(ctx, raw, authCtx.APIKey.PIIMode, endpoint) if err != nil { + err = clientGone(ctx, err) s.recordTerminalOutcome(metrics.OutcomeError, endpoint, execReq, &model, time.Since(start).Milliseconds()) + s.recordError(endpoint, execReq, &model, metrics.PhaseRequest, err) s.recordFailure(ctx, nil, &authCtx, endpoint, &execReq, &model, err, nil, time.Since(start).Milliseconds()) - return nil, err + return nil, reject(err, &model) } reservationID, err := s.metering.Reserve(ctx, authCtx, endpoint, execReq, model, uuid.NewString()) if err != nil { + err = clientGone(ctx, err) s.recordTerminalOutcome(metrics.OutcomeError, endpoint, execReq, &model, time.Since(start).Milliseconds()) - return nil, err + s.recordError(endpoint, execReq, &model, metrics.PhaseRequest, err) + return nil, reject(err, &model) } ctx, cancelInference := context.WithCancelCause(ctx) stopRenewal := func() {} if keeper, ok := s.metering.(ports.ReservationKeeper); ok { - stopRenewal = keeper.KeepReservation(ctx, reservationID, func() { cancelInference(domain.ErrInternal("credit reservation could not be maintained")) }) + stopRenewal = keeper.KeepReservation(ctx, reservationID, func() { + cancelInference(domain.ErrInternal("credit reservation could not be maintained").WithMeta("dependency", "metering", "cause", domain.CauseCreditHoldLost)) + }) } stopLease := func() { stopRenewal(); cancelInference(nil) } finish := func(target domain.PublicModel, o outcome) { @@ -98,7 +117,10 @@ func (s *GenerateService) Proxy(ctx context.Context, endpoint string, raw []byte } provider, err := s.registry.GetProvider(target.ProviderConfig.ProviderName) if err != nil { - return nil, domain.ErrProviderUnavailable(target.ProviderConfig.ProviderName) + // A provider name the catalog uses but this build doesn't know is + // Nexus's configuration, not the provider's outage. + return nil, domain.ErrProviderUnavailable(target.ProviderConfig.ProviderName).WithMeta( + "fault", domain.FaultNexus, "cause", domain.CauseProviderNotRegistered) } upstreamStart := time.Now() if !execReq.Stream { @@ -146,6 +168,7 @@ func (s *GenerateService) Proxy(ctx context.Context, endpoint string, raw []byte withProviderTarget(model, *model.Fallback).SupportsEndpoint(endpoint) { executed = withProviderTarget(model, *model.Fallback) s.logFallback(requestID, model, executed, sanitizeErrorWithPIIMapping(err, mapping)) + metrics.RecordFallback(endpoint, model.PublicModelID, model.ProviderConfig.ProviderName, executed.ProviderConfig.ProviderName, err) call, err = attempt(executed) } if err != nil { @@ -157,6 +180,7 @@ func (s *GenerateService) Proxy(ctx context.Context, endpoint string, raw []byte fields := s.buildErrorLogFields(ctx, requestID, &authCtx, endpoint, execReq.PublicModelID, executed, err, time.Since(start).Milliseconds()) s.logger.Error("provider request failed", fields...) s.recordGeneration(metrics.OutcomeError, endpoint, execReq, executed, time.Since(start).Milliseconds()) + s.recordError(endpoint, execReq, &executed, metrics.PhaseProvider, err) s.recordFailure(ctx, &reservationID, &authCtx, endpoint, &execReq, &executed, err, nil, time.Since(start).Milliseconds()) return nil, err } @@ -188,8 +212,14 @@ func (s *GenerateService) finishProxy(ctx context.Context, authCtx domain.AuthCo "chunks_received", o.chunks, "malformed_chunks", o.malformed, } if o.err != nil { - s.logger.Error("generation failed", append(fields, "error_type", fmt.Sprintf("%T", o.err))...) + fields = append(fields, "error_type", fmt.Sprintf("%T", o.err)) + s.logger.Error("generation failed", append(fields, domain.DiagnosticFields(o.err)...)...) s.recordGeneration(metrics.OutcomeError, endpoint, req, model, latencyMs) + phase := metrics.PhaseProvider + if req.Stream { + phase = metrics.PhaseStream + } + s.recordError(endpoint, req, &model, phase, o.err) s.recordFailure(ctx, &reservationID, &authCtx, endpoint, &req, &model, o.err, o.usage, latencyMs) return } diff --git a/go.mod b/go.mod index ed613ba..7f437c7 100644 --- a/go.mod +++ b/go.mod @@ -75,6 +75,7 @@ require ( github.com/in-toto/attestation v1.1.2 // indirect github.com/in-toto/in-toto-golang v0.9.0 // indirect github.com/jackc/pgx/v5 v5.8.0 // indirect + github.com/kylelemons/godebug v1.1.0 // indirect github.com/mdlayher/socket v0.6.0 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/oklog/ulid v1.3.1 // indirect diff --git a/pkg/buildinfo/buildinfo.go b/pkg/buildinfo/buildinfo.go new file mode 100644 index 0000000..6360a7d --- /dev/null +++ b/pkg/buildinfo/buildinfo.go @@ -0,0 +1,12 @@ +// Package buildinfo names the running build. Release images set both values +// at link time: +// +// go build -ldflags "-X github.com/dappnode/dappnode-nexus-gateway/pkg/buildinfo.Version=v0.1.67 -X github.com/dappnode/dappnode-nexus-gateway/pkg/buildinfo.Commit=" +package buildinfo + +var ( + // Version is the release tag, or "dev" for local builds. + Version = "dev" + // Commit is the source revision, when known. + Commit = "" +) diff --git a/pkg/domain/error_class.go b/pkg/domain/error_class.go new file mode 100644 index 0000000..e7dab64 --- /dev/null +++ b/pkg/domain/error_class.go @@ -0,0 +1,180 @@ +package domain + +import ( + "errors" + "strings" + "unicode/utf8" +) + +// Fault names who has to act on an error, so error reports can separate the +// caller's mistakes from provider outages and from Nexus's own failures. +const ( + FaultCaller = "caller" // The request or the account must change. + FaultProvider = "provider" // The upstream provider failed. + FaultNexus = "nexus" // Nexus must fix something: code, config or a provider account. + FaultUnclear = "unclear" // A provider refused the request: a bad request or a Nexus compatibility gap. +) + +// Error causes. A cause is a fixed value, never text from a request or a +// dependency, so it is always safe to log. +const ( + CausePIIFilterTimeout = "pii_filter_timeout" + CausePIIFilterFailed = "pii_filter_failed" + CauseProviderKeyMissing = "provider_key_missing" + CauseProviderNotRegistered = "provider_not_registered" + CauseCreditHoldLost = "credit_hold_lost" + CauseMeteringUnreachable = "metering_unreachable" + CauseMeteringErrorStatus = "metering_error_status" + CauseMeteringBadResponse = "metering_bad_response" + CauseRouterMissing = "router_not_configured" + CauseRouterEmptyModel = "router_empty_model" + CauseRouterUnreachable = "router_unreachable" + CauseRouterErrorStatus = "router_error_status" + CauseRouterBadResponse = "router_bad_response" + CauseReplayStoreFull = "replay_store_full" + CauseUntypedError = "untyped_error" +) + +var faultByCode = map[string]string{ + ErrCodeInvalidAPIKey: FaultCaller, + ErrCodeInactiveAPIKey: FaultCaller, + ErrCodeInactiveAccount: FaultCaller, + ErrCodeUnsupportedModel: FaultCaller, + ErrCodeUnsupportedEndpoint: FaultCaller, + ErrCodeUnsupportedFeature: FaultCaller, + ErrCodeContextLengthExceeded: FaultCaller, + ErrCodeToolSchemaInvalid: FaultCaller, + ErrCodeToolMessageInvalid: FaultCaller, + ErrCodeInvalidField: FaultCaller, + ErrCodeUnknownField: FaultCaller, + ErrCodeInsufficientBalance: FaultCaller, + ErrCodeClientCanceled: FaultCaller, + ErrCodeNotFound: FaultCaller, + ErrCodeForbidden: FaultCaller, + ErrCodeConflict: FaultCaller, + ErrCodeAlreadyExists: FaultCaller, + ErrCodeProviderRateLimited: FaultProvider, + ErrCodeProviderUnavailable: FaultProvider, + ErrCodeProviderTimeout: FaultProvider, + ErrCodeProviderError: FaultProvider, + ErrCodeProviderRejectedRequest: FaultUnclear, + ErrCodeProviderConfiguration: FaultNexus, + ErrCodeInternalError: FaultNexus, +} + +// FaultOf classifies an error. A "fault" in the metadata wins over the code, +// for codes that name a different party than the one at fault (a +// provider_unavailable raised by a Nexus dependency). Errors that are not +// GatewayErrors reach the client as internal errors, so they are Nexus's. +func FaultOf(err error) string { + var gwErr *GatewayError + if !errors.As(err, &gwErr) { + return FaultNexus + } + switch fault, _ := gwErr.Metadata["fault"].(string); fault { + case FaultCaller, FaultProvider, FaultNexus, FaultUnclear: + return fault + } + if fault, ok := faultByCode[gwErr.Code]; ok { + return fault + } + return FaultNexus +} + +// maxLogValue bounds a logged diagnostic string. +const maxLogValue = 128 + +// RedactedValue replaces an external value that is not identifier-shaped. +const RedactedValue = "[redacted]" + +// logSafeMeta lists metadata that is safe to log. Messages, reasons and +// anything that can echo request content stay out of logs. +var logSafeMeta = []struct { + key string + // external values come from callers, providers or other services and are + // logged only when they look like identifiers; the others are set by + // gateway code. + external bool +}{ + {"dependency", false}, {"cause", false}, {"contract", false}, + {"metering_status", false}, {"metering_code", true}, + {"router_status", false}, {"router_code", true}, + {"upstream_status", false}, {"upstream_type", true}, {"upstream_code", true}, + {"compatibility_note", false}, {"retry_reason", false}, {"attempt", false}, + {"account_id", true}, {"model", true}, {"provider", true}, {"requested_model", true}, +} + +// DiagnosticFields returns the error's classification and its log-safe +// metadata as key-value pairs, for the log line that reports the error. +func DiagnosticFields(err error) []any { + fields := []any{"fault", FaultOf(err)} + var gwErr *GatewayError + if !errors.As(err, &gwErr) { + return fields + } + fields = append(fields, "error_code", gwErr.Code, "gateway_status", gwErr.HTTPStatus) + for _, meta := range logSafeMeta { + switch value := gwErr.Metadata[meta.key].(type) { + case int, int64: + fields = append(fields, meta.key, value) + case string: + if value == "" { + continue + } + if meta.external && !isIdentifier(value) { + value = RedactedValue + } + fields = append(fields, meta.key, truncate(value, maxLogValue)) + } + } + return fields +} + +// isIdentifier reports whether s looks like a code, a model ID or an account +// ID rather than free text: short, with no spaces or punctuation beyond +// what identifiers use. +func isIdentifier(s string) bool { + if len(s) > maxLogValue { + return false + } + for _, r := range s { + switch { + case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9': + case strings.ContainsRune("-_./:@+", r): + default: + return false + } + } + return true +} + +// UpstreamStatus returns the provider's HTTP status, or 0 when there was none. +func UpstreamStatus(err error) int { + var gwErr *GatewayError + if errors.As(err, &gwErr) { + if status, ok := gwErr.Metadata["upstream_status"].(int); ok { + return status + } + } + return 0 +} + +// Annotate adds metadata to a GatewayError and leaves other errors unchanged. +func Annotate(err error, fields ...any) error { + var gwErr *GatewayError + if !errors.As(err, &gwErr) { + return err + } + return gwErr.WithMeta(fields...) +} + +func truncate(s string, n int) string { + if len(s) <= n { + return s + } + s = s[:n] + for !utf8.ValidString(s) { + s = s[:len(s)-1] + } + return s +} diff --git a/pkg/domain/error_class_test.go b/pkg/domain/error_class_test.go new file mode 100644 index 0000000..4c5019b --- /dev/null +++ b/pkg/domain/error_class_test.go @@ -0,0 +1,94 @@ +package domain + +import ( + "errors" + "strings" + "testing" +) + +func TestFaultOf(t *testing.T) { + for _, tc := range []struct { + err error + want string + }{ + {ErrUnsupportedModel("x"), FaultCaller}, + {ErrInsufficientBalance(), FaultCaller}, + {ErrClientCanceled(), FaultCaller}, + {ErrProviderUnavailable("p"), FaultProvider}, + {ErrProviderTimeout("p"), FaultProvider}, + {&GatewayError{HTTPStatus: 400, Code: ErrCodeProviderRejectedRequest}, FaultUnclear}, + {&GatewayError{HTTPStatus: 502, Code: ErrCodeProviderConfiguration}, FaultNexus}, + {ErrInternal("x"), FaultNexus}, + {errors.New("untyped"), FaultNexus}, + {&GatewayError{HTTPStatus: 500, Code: "something_new"}, FaultNexus}, + } { + if got := FaultOf(tc.err); got != tc.want { + t.Errorf("FaultOf(%v) = %q, want %q", tc.err, got, tc.want) + } + } +} + +// Only classification and log-safe metadata are logged: messages and free +// text that can echo a prompt never are. +func TestDiagnosticFieldsKeepOnlyLogSafeMetadata(t *testing.T) { + err := ErrInternal("an internal error occurred").WithMeta( + "dependency", "pii", + "cause", CausePIIFilterTimeout, + "upstream_status", 503, + "upstream_code", strings.Repeat("x", 300), + "upstream_type", "the prompt said: my password is hunter2", + "requested_model", "deepseek/deepseek-v4-flash", + "upstream_error", "the prompt said: my password is hunter2", + "reason", "free text", + "contract", "Tinfoil proof", + ) + got := map[string]any{} + fields := DiagnosticFields(err) + for i := 0; i+1 < len(fields); i += 2 { + got[fields[i].(string)] = fields[i+1] + } + for key, want := range map[string]any{ + "fault": FaultNexus, "error_code": ErrCodeInternalError, "gateway_status": 500, + "dependency": "pii", "cause": CausePIIFilterTimeout, "upstream_status": 503, + // External values are logged only when they look like identifiers. + "upstream_code": RedactedValue, "upstream_type": RedactedValue, + "requested_model": "deepseek/deepseek-v4-flash", + // Values set by gateway code are logged as they are. + "contract": "Tinfoil proof", + } { + if got[key] != want { + t.Errorf("%s = %v, want %v", key, got[key], want) + } + } + for _, key := range []string{"upstream_error", "reason"} { + if _, ok := got[key]; ok { + t.Errorf("%s was logged", key) + } + } +} + +// A fault in the metadata wins over the code: a provider_unavailable raised +// by a Nexus dependency is Nexus's. +func TestFaultOverride(t *testing.T) { + err := ErrProviderUnavailable("confidential replay protection").WithMeta("fault", FaultNexus) + if got := FaultOf(err); got != FaultNexus { + t.Fatalf("FaultOf = %q, want nexus", got) + } + if got := FaultOf(ErrProviderUnavailable("p").WithMeta("fault", "bogus")); got != FaultProvider { + t.Fatalf("an unknown override must be ignored, got %q", got) + } +} + +func TestUpstreamStatusAndAnnotate(t *testing.T) { + err := Annotate(ErrProviderUnavailable("p"), "upstream_status", 503) + if got := UpstreamStatus(err); got != 503 { + t.Fatalf("UpstreamStatus = %d", got) + } + if got := UpstreamStatus(errors.New("x")); got != 0 { + t.Fatalf("UpstreamStatus(untyped) = %d", got) + } + plain := errors.New("x") + if Annotate(plain, "a", 1) != plain { + t.Fatal("Annotate changed an untyped error") + } +} diff --git a/pkg/observability/logger/zap.go b/pkg/observability/logger/zap.go index 41371d6..39682af 100644 --- a/pkg/observability/logger/zap.go +++ b/pkg/observability/logger/zap.go @@ -4,6 +4,7 @@ import ( "fmt" "strings" + "github.com/dappnode/dappnode-nexus-gateway/pkg/buildinfo" "go.uber.org/zap" ) @@ -64,6 +65,8 @@ func requestLoggingConfig(level string) (zap.Config, bool) { // Request diagnostics are accounting evidence. Zap's production sampler // groups by message, so distinct request IDs can otherwise be dropped. cfg.Sampling = nil + // Every line names the build, so errors can be tied to a release. + cfg.InitialFields = map[string]any{"version": buildinfo.Version} cfg.Level = zap.NewAtomicLevel() verbose := false