fix(ops): Gemini 原生路由的网关本地错误按错误类型归类

Google 格式错误体没有 type 字段,ops 一律记为 api_error,客户端参数错误、
引用不存在的缓存等都被归为平台内部错误。

- googleError 写出时按状态码向 ops 登记错误类型:400/413 invalid_request_error、
  401 authentication_error、403 permission_error、404 not_found_error,
  429 与 5xx 不登记
- 计费、并发与风控拦截沿用 Claude / OpenAI 格式错误的类型与错误码,归类口径一致
- 存储价未配置、显式缓存未启用改用 googlePlatformError,仍按平台错误统计;
  错误透传规则改写的上游错误不登记本地类型
- 上游返回的缓存不存在标记为请求级失败:按客户端请求错误分类,
  上游原始错误保留为明细但不参与上游归因
This commit is contained in:
feeeei
2026-10-10 23:37:48 +08:00
parent d5a802f84a
commit 3abd646582
8 changed files with 252 additions and 18 deletions
@@ -42,7 +42,7 @@ func (h *GatewayHandler) geminiCachedContentsPreamble(c *gin.Context, component
return nil, middleware.AuthSubject{}, nil, false
}
if h.geminiCachedContentService == nil {
googleError(c, http.StatusNotFound, "Explicit context caching is not enabled")
googlePlatformError(c, http.StatusNotFound, "Explicit context caching is not enabled")
return nil, middleware.AuthSubject{}, nil, false
}
if apiKey.Group == nil || apiKey.GroupID == nil || apiKey.Group.Platform == service.PlatformComposite {
@@ -83,11 +83,11 @@ func readGeminiCachedContentBody(c *gin.Context) ([]byte, bool) {
func (h *GatewayHandler) checkGeminiCachedContentBilling(c *gin.Context, reqLog *zap.Logger, apiKey *service.APIKey, subscription *service.UserSubscription) bool {
if err := h.billingCacheService.CheckBillingEligibility(c.Request.Context(), apiKey.User, apiKey, apiKey.Group, subscription, service.QuotaPlatform(c.Request.Context(), apiKey)); err != nil {
reqLog.Info("gemini.cached_content.billing_eligibility_check_failed", zap.Error(err))
status, _, message, retryAfter := billingErrorDetails(err)
status, code, message, retryAfter := billingErrorDetails(err)
if retryAfter > 0 {
c.Header("Retry-After", strconv.Itoa(retryAfter))
}
googleError(c, status, message)
googleErrorWithType(c, status, code, "", message)
return false
}
return true
@@ -150,11 +150,11 @@ func (h *GatewayHandler) GeminiCachedContentsCreate(c *gin.Context) {
inflightRelease, err := reserveInflightBalance(c, h.billingCacheService, h.gatewayService, apiKey, subscription, tokenInflightEstimate(modelName, body))
if err != nil {
reqLog.Info("gemini.cached_content.inflight_reservation_rejected", zap.Error(err))
status, _, message, retryAfter := billingErrorDetails(err)
status, code, message, retryAfter := billingErrorDetails(err)
if retryAfter > 0 {
c.Header("Retry-After", strconv.Itoa(retryAfter))
}
googleError(c, status, message)
googleErrorWithType(c, status, code, "", message)
return
}
defer inflightRelease()
@@ -218,7 +218,7 @@ func (h *GatewayHandler) GeminiCachedContentsCreate(c *gin.Context) {
ChannelUsageFields: usageFields,
}) {
accountReleaseFunc()
googleError(c, http.StatusBadRequest, fmt.Sprintf("Explicit context caching is not available for model %s: storage pricing is not configured", modelName))
googlePlatformError(c, http.StatusBadRequest, fmt.Sprintf("Explicit context caching is not available for model %s: storage pricing is not configured", modelName))
return
}
upstreamModel, err := service.GeminiCachedContentUpstreamModelResource(account, mappedModel)
@@ -438,7 +438,7 @@ func (h *GatewayHandler) GeminiCachedContentsPatch(c *gin.Context) {
APIKey: apiKey, User: apiKey.User, Subscription: subscription, PricingAt: pricingAt,
}
if !h.gatewayService.CacheStoragePriced(c.Request.Context(), usage) {
googleError(c, http.StatusBadRequest, fmt.Sprintf("Explicit context caching is not available for model %s: storage pricing is not configured", record.Model))
googlePlatformError(c, http.StatusBadRequest, fmt.Sprintf("Explicit context caching is not available for model %s: storage pricing is not configured", record.Model))
return
}
if !h.checkGeminiCachedContentBilling(c, reqLog, apiKey, subscription) {
@@ -722,11 +722,11 @@ func (h *GatewayHandler) reserveGeminiCachedContentCharge(c *gin.Context, apiKey
}
func writeGeminiCachedContentBillingError(c *gin.Context, err error) {
status, _, message, retryAfter := billingErrorDetails(err)
status, code, message, retryAfter := billingErrorDetails(err)
if retryAfter > 0 {
c.Header("Retry-After", strconv.Itoa(retryAfter))
}
googleError(c, status, message)
googleErrorWithType(c, status, code, "", message)
}
// writeGeminiCachedContentUpstreamError 透传上游错误:缓存不存在统一返回标准 403(各上游形态不同且可能含上游资源 ID);
@@ -736,6 +736,7 @@ func writeGeminiCachedContentUpstreamError(c *gin.Context, upErr *service.Gemini
message := strings.TrimSpace(extractGoogleErrorMessage(upErr.Body))
service.SetOpsUpstreamError(c, status, message, "")
if upErr.NotFound() {
service.MarkOpsRequestScopedError(c, "permission_error")
c.Data(http.StatusForbidden, "application/json", []byte(service.GeminiCachedContentNotFoundResponse))
return
}
@@ -388,11 +388,11 @@ func (h *GatewayHandler) GeminiV1BetaModels(c *gin.Context) {
// 2) billing eligibility check (after wait)
if err := h.billingCacheService.CheckBillingEligibility(c.Request.Context(), apiKey.User, apiKey, apiKey.Group, subscription, service.QuotaPlatform(c.Request.Context(), apiKey)); err != nil {
reqLog.Info("gemini.billing_eligibility_check_failed", zap.Error(err))
status, _, message, retryAfter := billingErrorDetails(err)
status, code, message, retryAfter := billingErrorDetails(err)
if retryAfter > 0 {
c.Header("Retry-After", strconv.Itoa(retryAfter))
}
googleError(c, status, message)
googleErrorWithType(c, status, code, "", message)
return
}
@@ -400,11 +400,11 @@ func (h *GatewayHandler) GeminiV1BetaModels(c *gin.Context) {
inflightRelease, err := reserveInflightBalance(c, h.billingCacheService, h.gatewayService, apiKey, subscription, tokenInflightEstimate(modelName, body))
if err != nil {
reqLog.Info("gemini.inflight_reservation_rejected", zap.Error(err))
status, _, message, retryAfter := billingErrorDetails(err)
status, code, message, retryAfter := billingErrorDetails(err)
if retryAfter > 0 {
c.Header("Retry-After", strconv.Itoa(retryAfter))
}
googleError(c, status, message)
googleErrorWithType(c, status, code, "", message)
return
}
defer inflightRelease()
@@ -861,7 +861,8 @@ func (h *GatewayHandler) handleGeminiFailoverExhausted(c *gin.Context, failoverE
c.Set(service.OpsSkipPassthroughKey, true)
}
googleError(c, respCode, msg)
// 按透传规则改写的是上游错误,不登记本地错误类型。
googleErrorWithType(c, respCode, "", "", msg)
return
}
}
@@ -896,7 +897,21 @@ type pathParseError struct{ msg string }
func (e *pathParseError) Error() string { return e.msg }
// googleError 以 Google 错误格式回写网关产生的错误,并按状态码向 ops 登记错误类型:
// 4xx 为客户端请求问题,其余状态码不登记(按平台错误统计)。
func googleError(c *gin.Context, status int, message string) {
googleErrorWithType(c, status, googleClientErrorType(status), "", message)
}
// googlePlatformError 以 Google 错误格式回写由平台侧原因(如管理员未完成配置)导致的错误,ops 按平台错误统计。
func googlePlatformError(c *gin.Context, status int, message string) {
googleErrorWithType(c, status, "", "", message)
}
// googleErrorWithType 以 Google 错误格式回写错误,并向 ops 登记错误类型与错误码(Google 错误体没有 type 字段),
// 与 Claude / OpenAI 格式错误同口径分类;errType 为空时不登记。
func googleErrorWithType(c *gin.Context, status int, errType, code, message string) {
service.SetOpsLocalErrorType(c, errType, code)
c.JSON(status, gin.H{
"error": gin.H{
"code": status,
@@ -906,11 +921,27 @@ func googleError(c *gin.Context, status int, message string) {
})
}
// googleClientErrorType 返回客户端请求类 4xx 状态码对应的 ops 错误类型;其余状态码返回空串。
func googleClientErrorType(status int) string {
switch status {
case http.StatusBadRequest, http.StatusRequestEntityTooLarge:
return "invalid_request_error"
case http.StatusUnauthorized:
return "authentication_error"
case http.StatusForbidden:
return "permission_error"
case http.StatusNotFound:
return "not_found_error"
default:
return ""
}
}
// googleConcurrencyError 以 Google 错误格式回写并发槽获取失败,状态码与文案
// 沿用 concurrencyErrorResponse 的统一映射(客户端断开为 499)。
func googleConcurrencyError(c *gin.Context, err error, slotType string) {
status, _, _, message := concurrencyErrorResponse(err, slotType)
googleError(c, status, message)
status, errType, code, message := concurrencyErrorResponse(err, slotType)
googleErrorWithType(c, status, errType, code, message)
}
func writeUpstreamResponse(c *gin.Context, res *service.UpstreamHTTPResult) {
@@ -1183,6 +1183,14 @@ func OpsErrorLoggerMiddleware(ops *service.OpsService) gin.HandlerFunc {
}
}
// Google 格式错误体没有 type 字段:按网关回写时登记的错误类型补齐。
if errType, code := service.OpsLocalErrorType(c); errType != "" && !parsed.StreamFailure &&
(parsed.ErrorType == "" || parsed.ErrorType == "api_error") {
parsed.ErrorType = errType
if code != "" {
parsed.Code = code
}
}
normalizedType := normalizeOpsErrorType(parsed.ErrorType, parsed.Code)
phase, isBusinessLimited, errorOwner, errorSource := classifyOpsErrorLog(c, normalizedType, parsed.Message, parsed.Code, status)
@@ -2214,6 +2222,11 @@ func classifyOpsSeverity(errType string, status int) string {
func classifyOpsErrorLog(c *gin.Context, errType, message, code string, status int) (phase string, isBusinessLimited bool, errorOwner string, errorSource string) {
phase = classifyOpsPhase(errType, message, code)
if service.IsOpsRequestScopedError(c) {
// 失败由请求内容决定:只按错误类型分类,上游错误上下文不参与归因。
isBusinessLimited = classifyOpsIsBusinessLimited(errType, phase, code, status, message)
return phase, isBusinessLimited, classifyOpsErrorOwner(phase, message), classifyOpsErrorSource(phase, message)
}
routingCapacityLimited := isOpsRoutingCapacityLimited(c)
clientBusinessLimited := service.HasOpsClientBusinessLimited(c)
localModelConfiguration := clientBusinessLimited && service.OpsClientBusinessLimitedReason(c) == service.OpsClientBusinessLimitedReasonLocalModelConfiguration
@@ -0,0 +1,135 @@
package handler
import (
"net/http"
"net/http/httptest"
"testing"
"github.com/Wei-Shaw/sub2api/internal/service"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
)
// recordGoogleFormatOpsError 让 handle 以 Google 错误格式回写错误,返回 ops 中间件记录的错误日志。
func recordGoogleFormatOpsError(t *testing.T, handle gin.HandlerFunc) *service.OpsInsertErrorLogInput {
t.Helper()
setupOpsErrorLogTestQueue(t, 1)
gin.SetMode(gin.TestMode)
ops := service.NewOpsService(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)
router := gin.New()
router.Use(OpsErrorLoggerMiddleware(ops))
router.POST("/v1beta/cachedContents", handle)
router.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodPost, "/v1beta/cachedContents", nil))
require.Equal(t, int64(1), OpsErrorLogQueueLength())
return (<-opsErrorLogQueue).entry
}
func TestOpsErrorLoggerMiddleware_GoogleClientErrorsClassifiedAsClient(t *testing.T) {
for _, tc := range []struct {
status int
errType string
phase string
}{
{status: http.StatusBadRequest, errType: "invalid_request_error", phase: "request"},
{status: http.StatusRequestEntityTooLarge, errType: "invalid_request_error", phase: "request"},
{status: http.StatusUnauthorized, errType: "authentication_error", phase: "auth"},
{status: http.StatusForbidden, errType: "permission_error", phase: "request"},
{status: http.StatusNotFound, errType: "not_found_error", phase: "request"},
} {
t.Run(http.StatusText(tc.status), func(t *testing.T) {
entry := recordGoogleFormatOpsError(t, func(c *gin.Context) {
googleError(c, tc.status, "client request problem")
})
require.Equal(t, tc.status, entry.StatusCode)
require.Equal(t, tc.errType, entry.ErrorType)
require.Equal(t, tc.phase, entry.ErrorPhase)
require.Equal(t, "client", entry.ErrorOwner)
require.Equal(t, "client_request", entry.ErrorSource)
require.Equal(t, "P3", entry.Severity)
})
}
}
func TestOpsErrorLoggerMiddleware_GooglePlatformErrorStaysPlatform(t *testing.T) {
entry := recordGoogleFormatOpsError(t, func(c *gin.Context) {
googlePlatformError(c, http.StatusBadRequest, "Explicit context caching is not available for model m: storage pricing is not configured")
})
require.Equal(t, "api_error", entry.ErrorType)
require.Equal(t, "internal", entry.ErrorPhase)
require.Equal(t, "platform", entry.ErrorOwner)
require.Equal(t, "gateway", entry.ErrorSource)
}
func TestOpsErrorLoggerMiddleware_GoogleServerErrorsNotTypedByStatus(t *testing.T) {
entry := recordGoogleFormatOpsError(t, func(c *gin.Context) {
googleError(c, http.StatusInternalServerError, "Failed to persist cached content")
})
require.Equal(t, "api_error", entry.ErrorType)
require.Equal(t, "platform", entry.ErrorOwner)
}
func TestOpsErrorLoggerMiddleware_GoogleBillingErrorMatchesClaudeFormat(t *testing.T) {
entry := recordGoogleFormatOpsError(t, func(c *gin.Context) {
status, code, message, _ := billingErrorDetails(service.ErrInsufficientBalance)
googleErrorWithType(c, status, code, "", message)
})
require.Equal(t, "billing_error", entry.ErrorType)
require.Equal(t, "request", entry.ErrorPhase)
require.Equal(t, "client", entry.ErrorOwner)
require.True(t, entry.IsBusinessLimited)
}
func TestOpsErrorLoggerMiddleware_GoogleConcurrencyErrorClassifiedAsRequest(t *testing.T) {
entry := recordGoogleFormatOpsError(t, func(c *gin.Context) {
googleConcurrencyError(c, &ConcurrencyError{SlotType: "user"}, "user")
})
require.Equal(t, http.StatusTooManyRequests, entry.StatusCode)
require.Equal(t, "rate_limit_error", entry.ErrorType)
require.Equal(t, "request", entry.ErrorPhase)
require.Equal(t, "client", entry.ErrorOwner)
}
func TestOpsErrorLoggerMiddleware_GoogleErrorAfterUpstreamFailureKeepsUpstreamAttribution(t *testing.T) {
entry := recordGoogleFormatOpsError(t, func(c *gin.Context) {
service.SetOpsUpstreamError(c, http.StatusBadRequest, "upstream rejected", "")
googleError(c, http.StatusBadRequest, "upstream rejected")
})
require.Equal(t, "upstream", entry.ErrorPhase)
require.Equal(t, "provider", entry.ErrorOwner)
require.Equal(t, "upstream_http", entry.ErrorSource)
}
func TestOpsErrorLoggerMiddleware_RequestScopedUpstreamErrorClassifiedAsClient(t *testing.T) {
entry := recordGoogleFormatOpsError(t, func(c *gin.Context) {
service.SetOpsUpstreamError(c, http.StatusBadRequest, "Invalid resource state for cache content 8079767076522164224.", "")
c.Set(service.OpsUpstreamErrorsKey, []*service.OpsUpstreamErrorEvent{{
Platform: service.PlatformGemini, AccountID: 173, UpstreamStatusCode: http.StatusBadRequest, Kind: "http_error",
Message: "Invalid resource state for cache content 8079767076522164224.",
}})
service.MarkOpsRequestScopedError(c, "permission_error")
c.Data(http.StatusForbidden, "application/json", []byte(service.GeminiCachedContentNotFoundResponse))
})
require.Equal(t, http.StatusForbidden, entry.StatusCode)
require.Equal(t, "permission_error", entry.ErrorType)
require.Equal(t, "request", entry.ErrorPhase)
require.Equal(t, "client", entry.ErrorOwner)
require.Equal(t, "client_request", entry.ErrorSource)
require.False(t, entry.IsBusinessLimited)
require.NotNil(t, entry.UpstreamStatusCode, "上游原始错误保留为明细")
require.Equal(t, http.StatusBadRequest, *entry.UpstreamStatusCode)
}
func TestSetOpsLocalErrorTypeClearsRequestScopedMark(t *testing.T) {
gin.SetMode(gin.TestMode)
c, _ := gin.CreateTestContext(httptest.NewRecorder())
service.MarkOpsRequestScopedError(c, "permission_error")
require.True(t, service.IsOpsRequestScopedError(c))
service.SetOpsLocalErrorType(c, "", "")
require.False(t, service.IsOpsRequestScopedError(c))
errType, code := service.OpsLocalErrorType(c)
require.Empty(t, errType)
require.Empty(t, code)
}
@@ -100,9 +100,14 @@ func googleSecurityAuditError(c *gin.Context, decision *securityaudit.Decision)
return
}
if decision.Legacy != nil && decision.Legacy.Blocked {
googleError(c, securityAuditStatus(decision), securityAuditMessage(decision))
googleErrorWithType(c, securityAuditStatus(decision), securityAuditErrorCode(decision), "", securityAuditMessage(decision))
return
}
errType := "api_error"
if decision.Kind == securityaudit.DecisionBlock {
errType = "permission_error"
}
service.SetOpsLocalErrorType(c, errType, securityAuditErrorCode(decision))
status := securityAuditStatus(decision)
googleStatus := googleapi.HTTPStatusToGoogleStatus(status)
if status == http.StatusServiceUnavailable {
@@ -75,6 +75,7 @@ func TestGeminiForwardNative_CachedContentNotFoundPassthroughWithoutAccountPenal
require.Equal(t, 1, httpStub.calls, "不应重试")
require.Equal(t, http.StatusForbidden, rec.Code)
require.Equal(t, GeminiCachedContentNotFoundResponse, rec.Body.String(), "应返回标准缓存不存在响应")
require.True(t, IsOpsRequestScopedError(c), "缓存不存在按客户端请求错误统计")
})
}
}
@@ -177,6 +178,9 @@ func TestGeminiForwardNative_VertexCachedContentNotFoundReturnsStandardForbidden
require.Equal(t, http.StatusForbidden, rec.Code)
require.Equal(t, GeminiCachedContentNotFoundResponse, rec.Body.String())
require.NotContains(t, rec.Body.String(), "8079767076522164224", "不应暴露上游缓存 ID")
require.True(t, IsOpsRequestScopedError(c), "缓存不存在按客户端请求错误统计")
errType, _ := OpsLocalErrorType(c)
require.Equal(t, "permission_error", errType)
})
}
}
@@ -1834,10 +1834,11 @@ func (s *GeminiMessagesCompatService) writeGeminiNativeUpstreamError(c *gin.Cont
}
// writeGeminiCachedContentNotFound 记录上游原始错误,并向客户端统一返回缓存不存在的 403:
// 各上游的响应形态不同,且 Vertex 的文案含上游缓存资源 ID。
// 各上游的响应形态不同,且 Vertex 的文案含上游缓存资源 ID。失败由请求引用的缓存决定,ops 按客户端请求错误统计。
func (s *GeminiMessagesCompatService) writeGeminiCachedContentNotFound(c *gin.Context, account *Account, resp *http.Response, respBody []byte, requestID string, isOAuth bool) error {
respBody = unwrapIfNeeded(isOAuth, respBody)
upstreamMsg := s.recordGeminiNativeUpstreamError(c, account, resp.StatusCode, respBody, requestID)
MarkOpsRequestScopedError(c, "permission_error")
MarkResponseCommitted(c)
c.Data(http.StatusForbidden, "application/json", []byte(GeminiCachedContentNotFoundResponse))
return fmt.Errorf("gemini upstream error: %d message=%s", resp.StatusCode, upstreamMsg)
@@ -52,6 +52,14 @@ const (
// ensureForwardErrorResponse 检查此 key,为 true 时跳过兜底写入,避免在已完成的 JSON 后追加 SSE。
ResponseCommittedKey = "response_committed"
// OpsLocalErrorTypeKey / OpsLocalErrorCodeKey 记录网关以 Google 错误格式回写的错误类型与错误码:
// Google 错误体没有 type 字段,ops_error_logger 据此按与 Claude / OpenAI 格式错误相同的口径分类。
OpsLocalErrorTypeKey = "ops_local_error_type"
OpsLocalErrorCodeKey = "ops_local_error_code"
// OpsRequestScopedErrorKey 标记失败由请求内容决定(如引用的显式缓存不存在):只按错误类型分类,
// 上游错误上下文保留为明细但不参与归因。
OpsRequestScopedErrorKey = "ops_request_scoped_error"
OpsClientBusinessLimitedKey = "ops_client_business_limited"
OpsClientBusinessLimitedReasonKey = "ops_client_business_limited_reason"
OpsClientBusinessLimitedReasonIPRestriction = "api_key_ip_restriction"
@@ -134,6 +142,42 @@ func OpsClientBusinessLimitedReason(c *gin.Context) string {
return strings.TrimSpace(reason)
}
// SetOpsLocalErrorType 登记响应错误体缺少的错误类型与错误码,并清除请求级失败标记;errType 为空表示不登记。
func SetOpsLocalErrorType(c *gin.Context, errType, code string) {
if c == nil {
return
}
c.Set(OpsLocalErrorTypeKey, strings.TrimSpace(errType))
c.Set(OpsLocalErrorCodeKey, strings.TrimSpace(code))
c.Set(OpsRequestScopedErrorKey, false)
}
// OpsLocalErrorType 返回 SetOpsLocalErrorType 登记的错误类型与错误码;未登记时返回空串。
func OpsLocalErrorType(c *gin.Context) (errType, code string) {
if c == nil {
return "", ""
}
errType = c.GetString(OpsLocalErrorTypeKey)
if errType == "" {
return "", ""
}
return errType, c.GetString(OpsLocalErrorCodeKey)
}
// MarkOpsRequestScopedError 登记错误类型,并标记失败由请求内容决定:上游错误上下文不参与 ops 归因。
func MarkOpsRequestScopedError(c *gin.Context, errType string) {
if c == nil {
return
}
SetOpsLocalErrorType(c, errType, "")
c.Set(OpsRequestScopedErrorKey, true)
}
// IsOpsRequestScopedError 判断 MarkOpsRequestScopedError 是否标记了本次请求的失败。
func IsOpsRequestScopedError(c *gin.Context) bool {
return c != nil && c.GetBool(OpsRequestScopedErrorKey)
}
// OpsStreamError 描述承载在 2xx 响应上的带内错误:网关在响应状态已固化为 200 之后
// 就地以 SSE error 帧返回的错误(并发限流回退、Wait 后二次计费校验失败、流开始后才无
// 可用账号等),以及上游 2xx 正文或事件里携带的错误结果(NonStream 标记非流式正文)。