BCAI/easyai-ai-gateway
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(metrics): 补齐媒体结果存储读取与兼容响应指标

Browse Source 
增加对象存储 GET、结果 URL 签名和同步 Base64 限流指标，使生产轮询可直接验证零对象下载，并区分签名失败、容量等待超时和 20MiB 超限。\n\n验证：API 全量 go test、go vet、gofmt 与 git diff --check。
This commit is contained in:
wangbo
2026-08-05 18:55:41 +08:00
parent b13392ef50
commit 03c0873649
7 changed files with 113 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/internal/httpapi/compat_protocol_test.go
+1 -1
View File
@@ -276,7 +276,7 @@ func TestNativeOpenAIStreamPreservesUnknownEventFields(t *testing.T) {
writeProtocolCompatibleTaskResponse(
context.Background(), recorder, request, executor,
"chat.completions", "gpt-test", clients.ProtocolOpenAIChatCompletions,
store.GatewayTask{ID: "gateway-task"}, &auth.User{}, true, false,
store.GatewayTask{ID: "gateway-task"}, &auth.User{}, true, false, nil,
)
if recorder.Header().Get("X-Request-Id") != "req_stream_1" {
t.Fatalf("official stream header was lost: %+v", recorder.Header())
apps/api/internal/httpapi/handlers.go
+15 -3
View File
@@ -1315,7 +1315,7 @@ func (s *Server) createTask(kind string, compatible bool) http.Handler {
runCtx, cancelRun := s.requestExecutionContext(r)
defer cancelRun()
if responsePlan.compatibleMode {
writeProtocolCompatibleTaskResponse(runCtx, w, r, s.runner, kind, model, targetProtocol, task, user, responsePlan.streamMode, streamIncludeUsage(body))
writeProtocolCompatibleTaskResponse(runCtx, w, r, s.runner, kind, model, targetProtocol, task, user, responsePlan.streamMode, streamIncludeUsage(body), s.billingMetrics)
return
}
result, runErr := s.runner.Execute(runCtx, task, user)
@@ -1474,10 +1474,10 @@ type taskExecutor interface {
}
func writeCompatibleTaskResponse(runCtx context.Context, w http.ResponseWriter, r *http.Request, executor taskExecutor, kind string, model string, task store.GatewayTask, user *auth.User, streamMode bool, includeUsage bool) {
writeProtocolCompatibleTaskResponse(runCtx, w, r, executor, kind, model, "", task, user, streamMode, includeUsage)
writeProtocolCompatibleTaskResponse(runCtx, w, r, executor, kind, model, "", task, user, streamMode, includeUsage, nil)
}
func writeProtocolCompatibleTaskResponse(runCtx context.Context, w http.ResponseWriter, r *http.Request, executor taskExecutor, kind string, model string, targetProtocol string, task store.GatewayTask, user *auth.User, streamMode bool, includeUsage bool) {
func writeProtocolCompatibleTaskResponse(runCtx context.Context, w http.ResponseWriter, r *http.Request, executor taskExecutor, kind string, model string, targetProtocol string, task store.GatewayTask, user *auth.User, streamMode bool, includeUsage bool, base64Observer interface{ ObserveSynchronousBase64(string) }) {
if targetProtocol == "" || gatewayAPIV1Request(r) {
w.Header().Set("X-Gateway-Task-Id", task.ID)
}
@@ -1557,12 +1557,16 @@ func writeProtocolCompatibleTaskResponse(runCtx context.Context, w http.Response
releaseInlineResponse, limitErr := acquireSynchronousInlineResponseSlot(runCtx, kind, task.Request)
if limitErr != nil {
observeSynchronousBase64(base64Observer, "capacity_timeout")
writeProtocolError(w, targetProtocol, statusFromRunError(limitErr), runErrorMessage(limitErr), runErrorDetails(limitErr), runErrorCode(limitErr))
return
}
defer releaseInlineResponse()
result, runErr := executor.Execute(runCtx, task, user)
if runErr != nil {
if clients.ErrorCode(runErr) == "response_format_too_large" {
observeSynchronousBase64(base64Observer, "too_large")
}
if !requestStillConnected(r) {
return
}
@@ -1580,6 +1584,7 @@ func writeProtocolCompatibleTaskResponse(runCtx context.Context, w http.Response
if synchronousInlineResponseRequested(kind, task.Request) {
rawBytes := inlineMediaDecodedSize(result.Output)
if rawBytes > maxSynchronousInlineResponseBytes {
observeSynchronousBase64(base64Observer, "too_large")
writeProtocolError(w, targetProtocol, http.StatusRequestEntityTooLarge, "synchronous Base64 response exceeds the 20 MiB limit", map[string]any{
"task_id": result.Task.ID,
"query_url": "/api/v1/ai/result/" + result.Task.ID,
@@ -1590,6 +1595,7 @@ func writeProtocolCompatibleTaskResponse(runCtx context.Context, w http.Response
return
}
if rawBytes > 0 {
observeSynchronousBase64(base64Observer, "success")
w.Header().Set("X-Gateway-Response-Format", "b64_json")
} else {
w.Header().Set("X-Gateway-Response-Format", "url")
@@ -1620,6 +1626,12 @@ func acquireSynchronousInlineResponseSlot(ctx context.Context, kind string, requ
}
}
func observeSynchronousBase64(observer interface{ ObserveSynchronousBase64(string) }, outcome string) {
if observer != nil {
observer.ObserveSynchronousBase64(outcome)
}
}
func synchronousInlineResponseRequested(kind string, request map[string]any) bool {
if !mediaResultKind(kind) {
return false