fix(passthrough): emit UsageEvents (success + failure) with PK attribution

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::<UsageEvent>(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::<UsageEvent>(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"
+        );
+    }
 }