diff --git a/apps/api/internal/httpapi/compat_protocol_test.go b/apps/api/internal/httpapi/compat_protocol_test.go index 6bbeebe..d9d04f4 100644 --- a/apps/api/internal/httpapi/compat_protocol_test.go +++ b/apps/api/internal/httpapi/compat_protocol_test.go @@ -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()) diff --git a/apps/api/internal/httpapi/handlers.go b/apps/api/internal/httpapi/handlers.go index 54e9583..ead87b1 100644 --- a/apps/api/internal/httpapi/handlers.go +++ b/apps/api/internal/httpapi/handlers.go @@ -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 diff --git a/apps/api/internal/runner/object_storage_test.go b/apps/api/internal/runner/object_storage_test.go index 03bc446..ffdeaad 100644 --- a/apps/api/internal/runner/object_storage_test.go +++ b/apps/api/internal/runner/object_storage_test.go @@ -2,6 +2,7 @@ package runner import ( "context" + "fmt" "io" "net/http" "net/http/httptest" @@ -371,10 +372,16 @@ func TestObjectStorageReadUsesChannelRetryPolicy(t *testing.T) { if err != nil { t.Fatal(err) } - payload, err := readObjectStorageWithRetries(t.Context(), adapter, "media/request_asset/object.png") + var observed []string + payload, err := readObjectStorageWithRetriesObserved(t.Context(), adapter, "media/request_asset/object.png", func(event string, provider string, bytes int, _ time.Duration) { + observed = append(observed, fmt.Sprintf("%s:%s:%d", event, provider, bytes)) + }) if err != nil || string(payload) != "stored-payload" || calls.Load() != 2 { t.Fatalf("payload=%q calls=%d err=%v", payload, calls.Load(), err) } + if got := strings.Join(observed, ","); got != "read_failure:s3:0,read_success:s3:14" { + t.Fatalf("observed=%s", got) + } } func TestObjectStorageAuthFailureSkipsSameChannelRetry(t *testing.T) { diff --git a/apps/api/internal/runner/request_assets.go b/apps/api/internal/runner/request_assets.go index 4e0cade..e273587 100644 --- a/apps/api/internal/runner/request_assets.go +++ b/apps/api/internal/runner/request_assets.go @@ -292,7 +292,7 @@ func (s *Service) readRequestAssetBytes(ctx context.Context, asset store.Request if err != nil { return nil, err } - return readObjectStorageWithRetries(ctx, adapter, asset.ObjectKey) + return readObjectStorageWithRetriesObserved(ctx, adapter, asset.ObjectKey, s.observeObjectStorage) } if strings.TrimSpace(asset.LocalPath) != "" { payload, err := os.ReadFile(asset.LocalPath) @@ -331,10 +331,22 @@ func (s *Service) readRequestAssetBytes(ctx context.Context, asset store.Request } func readObjectStorageWithRetries(ctx context.Context, adapter *objectStorageAdapter, objectKey string) ([]byte, error) { + return readObjectStorageWithRetriesObserved(ctx, adapter, objectKey, nil) +} + +func readObjectStorageWithRetriesObserved(ctx context.Context, adapter *objectStorageAdapter, objectKey string, observe func(string, string, int, time.Duration)) ([]byte, error) { maxRetries, delays := uploadRetrySchedule(adapter.channel.RetryPolicy, adapter.channel.Provider) var lastErr error for attempt := 0; attempt <= maxRetries; attempt++ { + startedAt := time.Now() payload, err := adapter.get(ctx, objectKey) + if observe != nil { + event := "read_success" + if err != nil { + event = "read_failure" + } + observe(event, adapter.channel.Provider, len(payload), time.Since(startedAt)) + } if err == nil { return payload, nil } @@ -366,7 +378,13 @@ func (s *Service) requestAssetAccessURL(ctx context.Context, asset store.Request return value, nil } } - return adapter.presignGet(asset.ObjectKey, objectStorageSignedURLTTL) + value, err := adapter.presignGet(asset.ObjectKey, objectStorageSignedURLTTL) + if err != nil { + s.observeResultURLSigning("failure") + return "", err + } + s.observeResultURLSigning("success") + return value, nil } func (s *Service) fileStorageChannelForAsset(ctx context.Context, asset store.RequestAsset) (store.FileStorageChannel, error) { diff --git a/apps/api/internal/runner/service.go b/apps/api/internal/runner/service.go index 0744f46..7216fba 100644 --- a/apps/api/internal/runner/service.go +++ b/apps/api/internal/runner/service.go @@ -251,6 +251,13 @@ func (s *Service) observeResultStorage(source string) { } } +func (s *Service) observeResultURLSigning(outcome string) { + observer, ok := s.billingMetrics.(interface{ ObserveResultURLSigning(string) }) + if ok { + observer.ObserveResultURLSigning(outcome) + } +} + func (s *Service) Execute(ctx context.Context, task store.GatewayTask, user *auth.User) (Result, error) { return s.execute(ctx, task, user, nil) } diff --git a/apps/api/internal/securityevents/metrics.go b/apps/api/internal/securityevents/metrics.go index 0bc8748..8b5c781 100644 --- a/apps/api/internal/securityevents/metrics.go +++ b/apps/api/internal/securityevents/metrics.go @@ -109,11 +109,19 @@ type Metrics struct { storageChannelFailovers atomic.Uint64 storageAllChannelsFailed atomic.Uint64 storageObjectifiedBytes atomic.Uint64 + storageObjectReads atomic.Uint64 + storageObjectReadFailures atomic.Uint64 + storageObjectReadBytes atomic.Uint64 resultSourceUpstreamURL atomic.Uint64 resultSourceUploaded atomic.Uint64 resultPollResponses atomic.Uint64 resultPollResponseBytes atomic.Uint64 resultPollOversized atomic.Uint64 + resultURLSignSuccess atomic.Uint64 + resultURLSignFailure atomic.Uint64 + syncBase64Success atomic.Uint64 + syncBase64CapacityTimeout atomic.Uint64 + syncBase64TooLarge atomic.Uint64 taskAdmissionWaitBuckets [11]atomic.Uint64 taskAdmissionWaitMicros atomic.Uint64 } @@ -347,6 +355,13 @@ func (m *Metrics) ObserveObjectStorage(event string, provider string, bytes int6 m.storageChannelFailovers.Add(1) case "all_failed": m.storageAllChannelsFailed.Add(1) + case "read_success", "read_failure": + m.storageObjectReads.Add(1) + if event == "read_failure" { + m.storageObjectReadFailures.Add(1) + } else if bytes > 0 { + m.storageObjectReadBytes.Add(uint64(bytes)) + } } } @@ -371,6 +386,26 @@ func (m *Metrics) ObserveResultDelivery(event string, bytes int64) { } } +func (m *Metrics) ObserveResultURLSigning(outcome string) { + switch strings.ToLower(strings.TrimSpace(outcome)) { + case "success": + m.resultURLSignSuccess.Add(1) + case "failure": + m.resultURLSignFailure.Add(1) + } +} + +func (m *Metrics) ObserveSynchronousBase64(outcome string) { + switch strings.ToLower(strings.TrimSpace(outcome)) { + case "success": + m.syncBase64Success.Add(1) + case "capacity_timeout": + m.syncBase64CapacityTimeout.Add(1) + case "too_large": + m.syncBase64TooLarge.Add(1) + } +} + func (m *Metrics) ObserveTaskAdmissionWait(wait time.Duration) { if wait < 0 { wait = 0 @@ -587,6 +622,9 @@ func (m *Metrics) Handler(provider MetricsSnapshotProvider, issuer, audience str plainCounter(w, "easyai_gateway_storage_channel_failovers_total", "Switches to the next configured storage channel.", m.storageChannelFailovers.Load()) plainCounter(w, "easyai_gateway_storage_all_channels_failed_total", "Storage writes where every eligible channel failed.", m.storageAllChannelsFailed.Load()) plainCounter(w, "easyai_gateway_storage_objectified_bytes_total", "Binary bytes successfully written through storage channels.", m.storageObjectifiedBytes.Load()) + plainCounter(w, "easyai_gateway_storage_object_get_attempts_total", "Object storage GET attempts, including synchronous compatibility hydration.", m.storageObjectReads.Load()) + plainCounter(w, "easyai_gateway_storage_object_get_failures_total", "Failed object storage GET attempts.", m.storageObjectReadFailures.Load()) + plainCounter(w, "easyai_gateway_storage_object_get_bytes_total", "Bytes read from object storage GET responses.", m.storageObjectReadBytes.Load()) outcomeCounters(w, "easyai_gateway_result_storage_total", "Generated media results by canonical storage source.", []outcomeValue{ {"upstream_url", m.resultSourceUpstreamURL.Load()}, {"uploaded", m.resultSourceUploaded.Load()}, @@ -594,6 +632,15 @@ func (m *Metrics) Handler(provider MetricsSnapshotProvider, issuer, audience str plainCounter(w, "easyai_gateway_result_poll_responses_total", "Successful URL-only task result responses.", m.resultPollResponses.Load()) plainCounter(w, "easyai_gateway_result_poll_response_bytes_total", "Bytes written by successful URL-only task result responses.", m.resultPollResponseBytes.Load()) plainCounter(w, "easyai_gateway_result_poll_oversized_total", "Task result responses rejected by the URL-only size gate.", m.resultPollOversized.Load()) + outcomeCounters(w, "easyai_gateway_result_url_signing_total", "Private result URL signing attempts by outcome.", []outcomeValue{ + {"success", m.resultURLSignSuccess.Load()}, + {"failure", m.resultURLSignFailure.Load()}, + }) + outcomeCounters(w, "easyai_gateway_sync_base64_responses_total", "Synchronous Base64 compatibility responses by bounded outcome.", []outcomeValue{ + {"success", m.syncBase64Success.Load()}, + {"capacity_timeout", m.syncBase64CapacityTimeout.Load()}, + {"too_large", m.syncBase64TooLarge.Load()}, + }) publicErrorCounters(w, publicerror.MetricSnapshot()) platformModelRateLimitUtilizationGauges(w, modelRateLimits) plainGauge(w, "easyai_gateway_postgres_pool_max_connections", "Maximum PostgreSQL connections in this process pool.", int64(postgresPool.MaxConnections)) diff --git a/apps/api/internal/securityevents/metrics_test.go b/apps/api/internal/securityevents/metrics_test.go index 2041d71..224a2f7 100644 --- a/apps/api/internal/securityevents/metrics_test.go +++ b/apps/api/internal/securityevents/metrics_test.go @@ -57,10 +57,17 @@ func TestMetricsExposeBoundedOutcomesAndState(t *testing.T) { metrics.ObserveObjectStorage("retry", "s3", 0, 0) metrics.ObserveObjectStorage("failover", "s3", 0, 0) metrics.ObserveObjectStorage("all_failed", "", 0, 0) + metrics.ObserveObjectStorage("read_success", "aliyun_oss", 256, 15*time.Millisecond) + metrics.ObserveObjectStorage("read_failure", "aliyun_oss", 0, 5*time.Millisecond) metrics.ObserveResultStorage("upstream_url") metrics.ObserveResultStorage("uploaded") metrics.ObserveResultDelivery("poll_success", 512) metrics.ObserveResultDelivery("poll_oversized", 70<<10) + metrics.ObserveResultURLSigning("success") + metrics.ObserveResultURLSigning("failure") + metrics.ObserveSynchronousBase64("success") + metrics.ObserveSynchronousBase64("capacity_timeout") + metrics.ObserveSynchronousBase64("too_large") metrics.ObserveAsyncWorkerResize("success") metrics.ObserveConcurrencyLeaseRenewal("success") metrics.ObserveConcurrencyLeaseRenewal("lost") @@ -121,11 +128,19 @@ func TestMetricsExposeBoundedOutcomesAndState(t *testing.T) { `easyai_gateway_storage_channel_failovers_total 1`, `easyai_gateway_storage_all_channels_failed_total 1`, `easyai_gateway_storage_objectified_bytes_total 128`, + `easyai_gateway_storage_object_get_attempts_total 2`, + `easyai_gateway_storage_object_get_failures_total 1`, + `easyai_gateway_storage_object_get_bytes_total 256`, `easyai_gateway_result_storage_total{outcome="upstream_url"} 1`, `easyai_gateway_result_storage_total{outcome="uploaded"} 1`, `easyai_gateway_result_poll_responses_total 1`, `easyai_gateway_result_poll_response_bytes_total 512`, `easyai_gateway_result_poll_oversized_total 1`, + `easyai_gateway_result_url_signing_total{outcome="success"} 1`, + `easyai_gateway_result_url_signing_total{outcome="failure"} 1`, + `easyai_gateway_sync_base64_responses_total{outcome="success"} 1`, + `easyai_gateway_sync_base64_responses_total{outcome="capacity_timeout"} 1`, + `easyai_gateway_sync_base64_responses_total{outcome="too_large"} 1`, `easyai_gateway_platform_model_rate_limit_utilization{platform_model_id="platform-model-1",metric="concurrent"} 0.750000`, `easyai_gateway_async_worker_resizes_total{outcome="success"} 1`, `easyai_gateway_concurrency_lease_renewals_total{outcome="success"} 1`,