From 4b5722743e785a472bfdb4572d5c981f469fc30b Mon Sep 17 00:00:00 2001 From: Jarvis Date: Thu, 25 Jun 2026 18:12:03 +0800 Subject: [PATCH] fix(telemetry): emit a UsageEvent for failed non-chat requests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit chat / messages / responses emit a zero-token UsageEvent on a failed attempt (#655), so failures show up in the dashboard Logs and the budget ledger. The single-attempt non-chat handlers — /v1/completions, /v1/embeddings, /v1/rerank, /v1/audio/* and /v1/images/generations — dropped the event entirely on the error path, so a failed request was invisible: it appeared in neither Logs nor the ledger, only in metrics + the access log. Add a shared `usage_attr::emit_error_usage_event` helper and call it from each handler's error arm: one zero-token event carrying status_code, a bounded error_class (ProxyError::kind), the requested model name, api_key, and client IP/UA. model_id is left empty (the resolved id isn't threaded out of dispatch on the error path) — requested_model + status + error_class are enough to surface the row. The 501 NotImplemented path still emits nothing (no upstream call), unchanged. Two existing tests pinned the old "no event on 5xx" behavior (completions, rerank) — updated to assert the new zero-token error event. New equivalent tests added for embeddings, images and audio. Full aisix-proxy suite green (487), clippy + fmt clean. Origin: cross-API consistency audit after AISIX-Cloud#867. --- crates/aisix-proxy/src/audio.rs | 78 +++++++++++++++++++++++++++ crates/aisix-proxy/src/completions.rs | 48 ++++++++++++----- crates/aisix-proxy/src/embeddings.rs | 65 ++++++++++++++++++++++ crates/aisix-proxy/src/images.rs | 55 +++++++++++++++++++ crates/aisix-proxy/src/rerank.rs | 43 +++++++++++---- crates/aisix-proxy/src/usage_attr.rs | 45 ++++++++++++++++ 6 files changed, 311 insertions(+), 23 deletions(-) diff --git a/crates/aisix-proxy/src/audio.rs b/crates/aisix-proxy/src/audio.rs index ff3d2a4e..93696831 100644 --- a/crates/aisix-proxy/src/audio.rs +++ b/crates/aisix-proxy/src/audio.rs @@ -120,6 +120,19 @@ pub async fn transcriptions( RequestOutcome::from_status(status), elapsed, ); + // Per #655 parity: surface the failed request in Logs. The model + // isn't extracted from the multipart form on this error path, so + // requested_model is empty; status + error class still identify it. + crate::usage_attr::emit_error_usage_event( + &state, + "audio", + &request_id, + "", + &api_key_id, + status, + err.kind(), + &client, + ); err.into_response() } } @@ -193,6 +206,18 @@ pub async fn translations( RequestOutcome::from_status(status), elapsed, ); + // Per #655 parity: surface the failed request in Logs (model not + // extracted on the multipart error path → empty requested_model). + crate::usage_attr::emit_error_usage_event( + &state, + "audio", + &request_id, + "", + &api_key_id, + status, + err.kind(), + &client, + ); err.into_response() } } @@ -277,6 +302,18 @@ pub async fn speech( RequestOutcome::from_status(status), elapsed, ); + // Per #655 parity: surface the failed request in Logs with a + // zero-token event (status + error class). + crate::usage_attr::emit_error_usage_event( + &state, + "audio", + &request_id, + &model_name, + &api_key_id, + status, + err.kind(), + &client, + ); err.into_response() } } @@ -1250,4 +1287,45 @@ mod tests { }); assert_eq!(super::extract_token_usage(&v), None); } + + /// #655 parity: an upstream 5xx on /v1/audio/speech now emits ONE zero-token + /// UsageEvent so the failed request is visible in Logs (status + error + /// class) and attributed to the api_key — instead of being dropped. Mirrors + /// `completions.rs::upstream_5xx_emits_zero_token_error_event`. + #[tokio::test] + async fn speech_5xx_emits_zero_token_error_event() { + let upstream = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/v1/audio/speech")) + .respond_with(ResponseTemplate::new(500).set_body_string("Internal")) + .mount(&upstream) + .await; + + let snap = new_snap(&upstream.uri()); + snap.models.insert(tts_model("my-tts")); + snap.apikeys.insert(apikey_entry(&["*"])); + + let (tx, mut rx) = tokio::sync::mpsc::channel(8); + let app = build_app_with_sink(snap, tx); + let req = speech_req(r#"{"model":"my-tts","input":"hi","voice":"alloy"}"#); + let resp = tower::ServiceExt::oneshot(app, req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_GATEWAY); + + let ev = tokio::time::timeout(std::time::Duration::from_millis(500), rx.recv()) + .await + .expect("a failed /v1/audio/speech must emit a zero-token UsageEvent") + .expect("usage_sink sender dropped"); + assert_eq!(ev.status_code, 502, "upstream 5xx maps to 502"); + assert_eq!(ev.prompt_tokens, 0); + assert_eq!(ev.api_key_id, "k-1"); + assert_eq!(ev.requested_model, "my-tts"); + assert!( + !ev.error_class.is_empty(), + "error_class must classify the failure" + ); + assert!( + rx.try_recv().is_err(), + "exactly one event per failed request" + ); + } } diff --git a/crates/aisix-proxy/src/completions.rs b/crates/aisix-proxy/src/completions.rs index 0dd1f859..3559d453 100644 --- a/crates/aisix-proxy/src/completions.rs +++ b/crates/aisix-proxy/src/completions.rs @@ -143,6 +143,18 @@ pub async fn completions( RequestOutcome::from_status(status), elapsed, ); + // Per #655 parity: surface the failed request in Logs with a + // zero-token event (status + error class), instead of dropping it. + crate::usage_attr::emit_error_usage_event( + &state, + "completions", + &request_id, + &model_name, + &api_key_id, + status, + err.kind(), + &client, + ); err.into_response() } } @@ -830,13 +842,13 @@ mod tests { ); } - /// Issue #403 negative pinning: 4xx / 5xx responses must NOT - /// emit a UsageEvent. Audit MEDIUM-2 on PR #425 — a future - /// regression that moved `emit_usage_event` into the error - /// branch would silently ship without this kind of negative - /// assertion. + /// Per #655 parity (was #403 negative pinning): an upstream 5xx now emits + /// ONE zero-token UsageEvent so the failed request is visible in Logs + /// (status + error class) and attributed to the api_key — instead of being + /// dropped, as the non-chat handlers used to do. The 501 NotImplemented + /// path still emits nothing (no upstream call); see the test below. #[tokio::test] - async fn upstream_5xx_does_not_emit_usage_event() { + async fn upstream_5xx_emits_zero_token_error_event() { use aisix_obs::UsageSink; let upstream = MockServer::start().await; @@ -865,13 +877,23 @@ mod tests { .unwrap(); assert_eq!(resp.status(), StatusCode::BAD_GATEWAY); - let recv = tokio::time::timeout(std::time::Duration::from_millis(200), rx.recv()).await; - if let Ok(Some(ev)) = recv { - panic!( - "5xx must not emit UsageEvent, got status_code={}", - ev.status_code, - ); - } + let ev = tokio::time::timeout(std::time::Duration::from_millis(500), rx.recv()) + .await + .expect("a failed /v1/completions must emit a zero-token UsageEvent") + .expect("usage_sink sender dropped"); + assert_eq!(ev.status_code, 502, "upstream 5xx maps to 502"); + assert_eq!(ev.prompt_tokens, 0); + assert_eq!(ev.completion_tokens, 0); + assert_eq!(ev.api_key_id, "k-1"); + assert_eq!(ev.requested_model, "instruct"); + assert!( + !ev.error_class.is_empty(), + "error_class must classify the failure" + ); + assert!( + rx.try_recv().is_err(), + "exactly one event per failed request" + ); } /// Issue #403 audit MEDIUM-3: the 501 NotImplemented path diff --git a/crates/aisix-proxy/src/embeddings.rs b/crates/aisix-proxy/src/embeddings.rs index 30428eb1..23f789dd 100644 --- a/crates/aisix-proxy/src/embeddings.rs +++ b/crates/aisix-proxy/src/embeddings.rs @@ -192,6 +192,18 @@ pub async fn embeddings( RequestOutcome::from_status(status), elapsed, ); + // Per #655 parity: surface the failed request in Logs with a + // zero-token event (status + error class), instead of dropping it. + crate::usage_attr::emit_error_usage_event( + &state, + "embeddings", + &request_id, + &model_name, + &api_key_id, + status, + err.kind(), + &client, + ); err.into_response() } } @@ -1346,6 +1358,59 @@ mod tests { ); } + /// #655 parity: an upstream 5xx on /v1/embeddings now emits ONE zero-token + /// UsageEvent so the failed request is visible in Logs (status + error + /// class) and attributed to the api_key — instead of being dropped, as the + /// non-chat handlers used to do. Mirrors + /// `completions.rs::upstream_5xx_emits_zero_token_error_event`. + #[tokio::test] + async fn upstream_5xx_emits_zero_token_error_event() { + use aisix_obs::UsageSink; + + let upstream = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/embeddings")) + .respond_with(ResponseTemplate::new(500).set_body_string("Internal")) + .mount(&upstream) + .await; + + let snap = new_snap(&upstream.uri()); + snap.models.insert(model_entry("my-embed")); + snap.apikeys.insert(apikey_entry(&["*"])); + + let (tx, mut rx) = tokio::sync::mpsc::channel(8); + let hub = Arc::new(Hub::new()); + hub.register_specialized("openai", Arc::new(OpenAiBridge::new())); + let handle = SnapshotHandle::new(snap); + let state = crate::ProxyState::new(handle, hub, &cfg()) + .without_cache() + .with_usage_sink(UsageSink::new(tx)); + let app = crate::build_router(state); + + let body = serde_json::json!({"model": "my-embed", "input": "hi"}); + let resp = tower::ServiceExt::oneshot(app, make_req(body)) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_GATEWAY); + + let ev = tokio::time::timeout(std::time::Duration::from_millis(500), rx.recv()) + .await + .expect("a failed /v1/embeddings must emit a zero-token UsageEvent") + .expect("usage_sink sender dropped"); + assert_eq!(ev.status_code, 502, "upstream 5xx maps to 502"); + assert_eq!(ev.prompt_tokens, 0); + assert_eq!(ev.api_key_id, "k-1"); + assert_eq!(ev.requested_model, "my-embed"); + assert!( + !ev.error_class.is_empty(), + "error_class must classify the failure" + ); + assert!( + rx.try_recv().is_err(), + "exactly one event per failed request" + ); + } + /// Malformed JSON (syntax error) on /v1/embeddings must also /// surface as 400, not 422. Same JsonRejection → InvalidRequest /// path as the missing-field cases. diff --git a/crates/aisix-proxy/src/images.rs b/crates/aisix-proxy/src/images.rs index 4bd61fd4..0f09b1bf 100644 --- a/crates/aisix-proxy/src/images.rs +++ b/crates/aisix-proxy/src/images.rs @@ -127,6 +127,18 @@ pub async fn image_generations( RequestOutcome::from_status(status), elapsed, ); + // Per #655 parity: surface the failed request in Logs with a + // zero-token event (status + error class), instead of dropping it. + crate::usage_attr::emit_error_usage_event( + &state, + "images", + &request_id, + &model_name, + &api_key_id, + status, + err.kind(), + &client, + ); err.into_response() } } @@ -911,4 +923,47 @@ mod tests { "pk_label must mirror telemetry_tags.pk_label" ); } + + /// #655 parity: an upstream 5xx on /v1/images/generations now emits ONE + /// zero-token UsageEvent so the failed request is visible in Logs (status + + /// error class) and attributed to the api_key — instead of being dropped. + /// Mirrors `completions.rs::upstream_5xx_emits_zero_token_error_event`. + #[tokio::test] + async fn upstream_5xx_emits_zero_token_error_event() { + let upstream = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/images/generations")) + .respond_with(ResponseTemplate::new(500).set_body_string("Internal")) + .mount(&upstream) + .await; + + let snap = new_snap(&upstream.uri()); + snap.models.insert(model_entry("dall-e")); + snap.apikeys.insert(apikey_entry(&["*"])); + + let (tx, mut rx) = tokio::sync::mpsc::channel(8); + let app = build_app_with_sink(snap, tx); + let req = serde_json::json!({"model": "dall-e", "prompt": "a cat", "n": 1}); + let resp = tower::ServiceExt::oneshot(app, make_req(req)) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_GATEWAY); + + let ev = tokio::time::timeout(std::time::Duration::from_millis(500), rx.recv()) + .await + .expect("a failed /v1/images/generations must emit a zero-token UsageEvent") + .expect("usage_sink sender dropped"); + assert_eq!(ev.status_code, 502, "upstream 5xx maps to 502"); + assert_eq!(ev.prompt_tokens, 0); + assert_eq!(ev.api_key_id, "k-1"); + assert_eq!(ev.requested_model, "dall-e"); + assert!( + !ev.error_class.is_empty(), + "error_class must classify the failure" + ); + assert!( + rx.try_recv().is_err(), + "exactly one event per failed request" + ); + } } diff --git a/crates/aisix-proxy/src/rerank.rs b/crates/aisix-proxy/src/rerank.rs index 75900a88..c9f0e57f 100644 --- a/crates/aisix-proxy/src/rerank.rs +++ b/crates/aisix-proxy/src/rerank.rs @@ -131,6 +131,18 @@ pub async fn rerank( RequestOutcome::from_status(status), elapsed, ); + // Per #655 parity: surface the failed request in Logs with a + // zero-token event (status + error class), instead of dropping it. + crate::usage_attr::emit_error_usage_event( + &state, + "rerank", + &request_id, + &model_name, + &api_key_id, + status, + err.kind(), + &client, + ); err.into_response() } } @@ -1309,10 +1321,12 @@ mod tests { } } - /// Issue #405 negative pinning: upstream 5xx must NOT emit - /// a UsageEvent (same discipline as #425 audit MEDIUM-2). + /// Per #655 parity (was #405 negative pinning): an upstream 5xx now emits + /// ONE zero-token UsageEvent so the failed /v1/rerank request is visible in + /// Logs (status + error class), instead of being dropped. The 200-without- + /// usage-fields case (test above) still emits nothing. #[tokio::test] - async fn upstream_5xx_does_not_emit_usage_event() { + async fn upstream_5xx_emits_zero_token_error_event() { use aisix_obs::UsageSink; let upstream = MockServer::start().await; @@ -1345,13 +1359,22 @@ mod tests { .unwrap(); assert_eq!(resp.status(), StatusCode::BAD_GATEWAY); - let recv = tokio::time::timeout(std::time::Duration::from_millis(200), rx.recv()).await; - if let Ok(Some(ev)) = recv { - panic!( - "5xx must not emit UsageEvent, got status_code={}", - ev.status_code, - ); - } + let ev = tokio::time::timeout(std::time::Duration::from_millis(500), rx.recv()) + .await + .expect("a failed /v1/rerank must emit a zero-token UsageEvent") + .expect("usage_sink sender dropped"); + assert_eq!(ev.status_code, 502, "upstream 5xx maps to 502"); + assert_eq!(ev.prompt_tokens, 0); + assert_eq!(ev.requested_model, "rerank-openai"); + assert_eq!(ev.api_key_id, "k-1"); + assert!( + !ev.error_class.is_empty(), + "error_class must classify the failure" + ); + assert!( + rx.try_recv().is_err(), + "exactly one event per failed request" + ); } /// AISIX-Cloud#867 parity: a successful /v1/rerank 200 must stamp the diff --git a/crates/aisix-proxy/src/usage_attr.rs b/crates/aisix-proxy/src/usage_attr.rs index 3a7cfd5a..45b6a58d 100644 --- a/crates/aisix-proxy/src/usage_attr.rs +++ b/crates/aisix-proxy/src/usage_attr.rs @@ -13,6 +13,8 @@ use aisix_core::AisixSnapshot; use aisix_obs::UsageEvent; use crate::chat::sanitize_tag; +use crate::client_ip::ClientContext; +use crate::state::ProxyState; /// Resolve a ProviderKey's telemetry attribution tags from the live snapshot. /// An empty `provider_key_id` (pre-dispatch error paths) or an id with no @@ -48,3 +50,46 @@ pub(crate) fn apply_pk_telemetry( event.pk_label = sanitize_tag(tags.pk_label.unwrap_or_default()); event.byo_label = sanitize_tag(tags.byo_label.unwrap_or_default()); } + +/// Emit ONE zero-token `UsageEvent` for a FAILED request on a non-chat handler +/// (completions / embeddings / rerank / audio / images), so the dashboard Logs +/// and budget ledger surface the failure (status and bounded error class) +/// instead of dropping it. Mirrors the #655 behavior chat / messages / +/// responses already have: those endpoints emit a zero-token event per failed +/// attempt; the single-attempt non-chat handlers emit one terminal event here. +/// +/// `model_id` is intentionally left empty — on the error path the resolved +/// Model id isn't threaded back out of dispatch, but `requested_model`, +/// `api_key_id`, `status_code` and `error_class` are enough for the request to +/// appear in Logs. `label` is the usage_sink bucket (#408); all five callers +/// are OpenAI-shaped, so `inbound_protocol` is `"openai"`. +#[allow(clippy::too_many_arguments)] +pub(crate) fn emit_error_usage_event( + state: &ProxyState, + label: &'static str, + request_id: &str, + requested_model: &str, + api_key_id: &str, + status_code: u16, + error_class: &str, + client: &ClientContext, +) { + let event = UsageEvent { + request_id: request_id.to_string(), + occurred_at: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true), + api_key_id: api_key_id.to_string(), + requested_model: requested_model.to_string(), + status_code, + inbound_protocol: "openai".to_string(), + error_class: error_class.to_string(), + client_source_ip: client.source_ip.clone(), + client_user_agent: client.user_agent.clone(), + ..Default::default() + }; + state.usage_sink.try_emit(label, event.clone()); + let snap = state.snapshot.load(); + let exporters = snap.observability_exporters.entries(); + state + .otlp_fan_out + .fan_out(&event, None, exporters.iter().map(|e| &e.value)); +}