diff --git a/crates/aisix-proxy/src/passthrough.rs b/crates/aisix-proxy/src/passthrough.rs index 89fb0b71..328e5833 100644 --- a/crates/aisix-proxy/src/passthrough.rs +++ b/crates/aisix-proxy/src/passthrough.rs @@ -152,7 +152,7 @@ pub async fn passthrough( ) .await { - Ok((resp, provider_label)) => { + Ok((resp, provider_label, provider_key_id)) => { let elapsed = started.elapsed(); let status = resp.status().as_u16(); emit_access_log( @@ -175,6 +175,19 @@ pub async fn passthrough( RequestOutcome::from_status(status), elapsed, ); + // #699: record the passthrough call in the UsageEvent stream — + // pre-fix it never appeared in /logs, the budget ledger or the + // exporters. The raw tunnel parses nothing, so tokens stay zero; + // the upstream's status is relayed verbatim and recorded as-is. + emit_usage_event( + &state, + &request_id, + &api_key_id, + &provider_key_id, + status, + elapsed, + &client, + ); resp } Err(err) => { @@ -196,6 +209,20 @@ pub async fn passthrough( RequestOutcome::from_status(status), elapsed, ); + // #699 / #655 parity: surface the failed request in Logs with a + // zero-token event (status + error class). No resolved model on + // the error path -> empty requested_model, like audio's multipart + // error path. + crate::usage_attr::emit_error_usage_event( + &state, + "passthrough", + &request_id, + "", + &api_key_id, + status, + err.kind(), + &client, + ); err.into_response() } } @@ -210,7 +237,7 @@ async fn dispatch( req: Request, request_id: &str, source_ip: &str, -) -> Result<(Response, String), ProxyError> { +) -> Result<(Response, String, String), ProxyError> { let snapshot = state.snapshot.load(); // Find a model for this provider so we can borrow its provider_key. @@ -507,7 +534,42 @@ async fn dispatch( ); } - Ok((response, provider_lower)) + Ok((response, provider_lower, pk_entry.id.to_string())) +} + +/// #699: push one zero-token `UsageEvent` per passthrough request onto the +/// CP sink and the exporter fan-out. The raw tunnel parses neither request +/// nor response, so there are no tokens or model fields — the event records +/// who lent which provider's credentials (per-PK attribution tags), the +/// relayed status, and the latency. `inbound_protocol = "passthrough"` +/// distinguishes these rows from typed-endpoint traffic in /logs. +fn emit_usage_event( + state: &ProxyState, + request_id: &str, + api_key_id: &str, + provider_key_id: &str, + status_code: u16, + elapsed: Duration, + client: &crate::client_ip::ClientContext, +) { + let snap = state.snapshot.load(); + let mut event = aisix_obs::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(), + status_code, + latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32, + inbound_protocol: "passthrough".to_string(), + client_source_ip: client.source_ip.clone(), + client_user_agent: client.user_agent.clone(), + ..Default::default() + }; + crate::usage_attr::apply_pk_telemetry(&mut event, &snap, provider_key_id); + state.usage_sink.try_emit("passthrough", event.clone()); + let exporters = snap.observability_exporters.entries(); + state + .otlp_fan_out + .fan_out(&event, None, exporters.iter().map(|e| &e.value)); } /// Copy response headers that are safe to relay to the downstream caller. @@ -1311,4 +1373,100 @@ mod tests { let resp = ServiceExt::oneshot(app, req).await.unwrap(); assert_eq!(resp.status(), StatusCode::OK); } + + /// #699: a passthrough request must land in the UsageEvent stream — + /// zero tokens (nothing is parsed), the relayed upstream status, the + /// caller's api_key and the lent ProviderKey's attribution tags. + #[tokio::test] + async fn emits_usage_event_on_success_issue_699() { + use aisix_obs::{UsageEvent, UsageSink}; + + let upstream = MockServer::start().await; + Mock::given(wm_method("GET")) + .and(path("/v1/models")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"data": []}))) + .mount(&upstream) + .await; + + let snap = AisixSnapshot::new(); + let pk_json = format!( + r#"{{"display_name":"openai-up","secret":"sk-test","api_base":"{}","provider":"openai","adapter":"openai","telemetry_tags":{{"kind":"catalog","featured":true,"branded_provider":"openai","pk_label":"prod-pt-key"}}}}"#, + upstream.uri() + ); + let pk: aisix_core::ProviderKey = serde_json::from_str(&pk_json).unwrap(); + snap.provider_keys.insert(ResourceEntry::new(PK_ID, pk, 1)); + snap.models.insert(openai_model("gpt")); + snap.apikeys.insert(apikey_entry(&["*"])); + + let (tx, mut rx) = tokio::sync::mpsc::channel::(8); + let hub = Arc::new(Hub::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 req = Request::builder() + .method("GET") + .uri("/passthrough/openai/v1/models") + .header("authorization", "Bearer sk-caller") + .body(axum::body::Body::empty()) + .unwrap(); + let resp = ServiceExt::oneshot(app, req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + + let event = tokio::time::timeout(std::time::Duration::from_millis(500), rx.recv()) + .await + .expect("a passthrough request must emit a UsageEvent") + .expect("usage_sink sender dropped"); + assert_eq!(event.inbound_protocol, "passthrough"); + assert_eq!(event.status_code, 200); + assert_eq!(event.api_key_id, "k-1"); + assert_eq!(event.prompt_tokens, 0, "the raw tunnel parses no tokens"); + assert_eq!(event.pk_label, "prod-pt-key", "PK attribution must apply"); + assert!( + rx.try_recv().is_err(), + "exactly one event per passthrough request" + ); + } + + /// #699 / #655 parity: a failed passthrough dispatch (no model for the + /// provider -> 404) emits one zero-token error event instead of being + /// dropped. + #[tokio::test] + async fn failed_dispatch_emits_zero_token_error_event_issue_699() { + use aisix_obs::{UsageEvent, UsageSink}; + + let snap = new_snap("http://unused"); + snap.models.insert(openai_model("gpt")); + snap.apikeys.insert(apikey_entry(&["*"])); + + let (tx, mut rx) = tokio::sync::mpsc::channel::(8); + let hub = Arc::new(Hub::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 req = Request::builder() + .method("GET") + .uri("/passthrough/nonexistent/v1/models") + .header("authorization", "Bearer sk-caller") + .body(axum::body::Body::empty()) + .unwrap(); + let resp = ServiceExt::oneshot(app, req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::NOT_FOUND); + + let event = tokio::time::timeout(std::time::Duration::from_millis(500), rx.recv()) + .await + .expect("a failed passthrough must emit a zero-token UsageEvent") + .expect("usage_sink sender dropped"); + assert_eq!(event.status_code, 404); + assert_eq!(event.api_key_id, "k-1"); + assert!( + !event.error_class.is_empty(), + "error_class must classify the failure" + ); + } }