feat(telemetry): split upstream and downstream latency on UsageEvent

crates/aisix-obs/src/otlp_http_sink.rs

@@ -592,8 +592,11 @@ fn build_otlp_span(record: &SinkRecord, exporter_name: &str) -> Value {
     // "now" so the span isn't silently dropped.
     let end_unix_nano =
         parse_rfc3339_to_unix_nano(&event.occurred_at).unwrap_or_else(now_unix_nano);
+    // The span represents this ATTEMPT, so its duration is the
+    // attempt-scoped upstream latency (the request-scoped
+    // `downstream_latency_ms` rides along as an attribute instead).
     // Latency landed in milliseconds; widen + multiply.
-    let latency_nanos = (event.latency_ms as u128).saturating_mul(1_000_000);
+    let latency_nanos = (event.upstream_latency_ms as u128).saturating_mul(1_000_000);
     let start_unix_nano = end_unix_nano.saturating_sub(latency_nanos);
 
     // Status: OK (1) for 2xx, ERROR (2) otherwise.
@@ -653,8 +656,20 @@ fn build_otlp_span(record: &SinkRecord, exporter_name: &str) -> Value {
     }
     attributes.push(attr_string("aisix.exporter_name", exporter_name));
     attributes.push(attr_string("aisix.request_id", &event.request_id));
-    if event.ttft_ms > 0 {
-        attributes.push(attr_int("aisix.ttft_ms", event.ttft_ms as i64));
+    if event.upstream_ttft_ms > 0 {
+        attributes.push(attr_int(
+            "aisix.upstream_ttft_ms",
+            event.upstream_ttft_ms as i64,
+        ));
+    }
+    // Request-scoped: present only on the attempt that delivered the
+    // terminal response, so consumers read a request's caller-facing
+    // latency off that one span rather than summing the group.
+    if event.downstream_latency_ms > 0 {
+        attributes.push(attr_int(
+            "aisix.downstream_latency_ms",
+            event.downstream_latency_ms as i64,
+        ));
     }
     // Per-attempt telemetry (#655). `request_id` is the trace/group key; a
     // failover request emits one span per attempt sharing it, ordered by
@@ -892,7 +907,7 @@ mod tests {
             api_key_id: "ak-uuid".into(),
             prompt_tokens: 10,
             completion_tokens: 5,
-            latency_ms: 250,
+            upstream_latency_ms: 250,
             status_code: 200,
             provider_request_id: "chatcmpl-abc".into(),
             provider_model_version: "gpt-4o-2024-08-06".into(),
@@ -1468,19 +1483,22 @@ mod tests {
         assert!(!keys.contains(&"gen_ai.response.model"));
         assert!(!keys.contains(&"gen_ai.response.finish_reasons"));
         // ttft_ms = 0 (default) → omitted
-        assert!(!keys.contains(&"aisix.ttft_ms"));
+        assert!(!keys.contains(&"aisix.upstream_ttft_ms"));
     }
 
     #[test]
     fn payload_includes_ttft_when_set() {
         let mut ev = sample_event();
-        ev.ttft_ms = 42;
+        ev.upstream_ttft_ms = 42;
         let body = build_otlp_traces_payload(&ev, "test-exp");
         let attrs = body["resourceSpans"][0]["scopeSpans"][0]["spans"][0]["attributes"]
             .as_array()
             .unwrap();
-        let ttft_attr = attrs.iter().find(|a| a["key"] == "aisix.ttft_ms");
-        assert!(ttft_attr.is_some(), "aisix.ttft_ms should be present");
+        let ttft_attr = attrs.iter().find(|a| a["key"] == "aisix.upstream_ttft_ms");
+        assert!(
+            ttft_attr.is_some(),
+            "aisix.upstream_ttft_ms should be present"
+        );
         assert_eq!(ttft_attr.unwrap()["value"]["intValue"], "42");
     }
 

crates/aisix-obs/src/sink/datadog.rs

@@ -474,7 +474,7 @@ mod tests {
             status_code: 200,
             prompt_tokens: 5,
             completion_tokens: 7,
-            latency_ms: 123,
+            upstream_latency_ms: 123,
             provider_model_version: "gpt-4o-2024-08-06".into(),
             finish_reason: "stop".into(),
             ..UsageEvent::default()
@@ -523,7 +523,7 @@ mod tests {
         // AISIX custom dimensions under the `aisix.` prefix.
         assert_eq!(log["aisix.request_id"], "req-42");
         assert_eq!(log["aisix.model_id"], "gpt-4o");
-        assert_eq!(log["aisix.latency_ms"], 123);
+        assert_eq!(log["aisix.upstream_latency_ms"], 123);
 
         // The API key must NEVER appear in the body anywhere.
         let body_text = serde_json::to_string(&logs).unwrap();

crates/aisix-obs/src/sink/sls.rs

@@ -487,7 +487,7 @@ mod tests {
             status_code: 200,
             prompt_tokens: 5,
             completion_tokens: 7,
-            latency_ms: 123,
+            upstream_latency_ms: 123,
             ..UsageEvent::default()
         };
         let ack = sink
@@ -559,7 +559,10 @@ mod tests {
             contents.get("completion_tokens").map(String::as_str),
             Some("7")
         );
-        assert_eq!(contents.get("latency_ms").map(String::as_str), Some("123"));
+        assert_eq!(
+            contents.get("upstream_latency_ms").map(String::as_str),
+            Some("123")
+        );
         // Empty metadata is omitted uniformly: `api_key_id` (serde `default`
         // only, would serialize as "") and `finish_reason` (`skip_serializing_if`)
         // both drop out, so the SLS log carries no blank columns.

crates/aisix-obs/src/usage.rs

@@ -112,14 +112,57 @@ pub struct UsageEvent {
     #[serde(default, skip_serializing_if = "is_false")]
     pub usage_estimated: bool,
 
-    pub latency_ms: u32,
-
-    /// Time to first token in milliseconds. Only meaningful on the
-    /// streaming path — measures elapsed time from request entry to
-    /// the first upstream SSE chunk. 0 on non-streaming, error, and
+    /// How long THIS attempt spent on the upstream, in milliseconds:
+    /// from the moment the attempt began to the moment it settled —
+    /// end-of-stream for a streamed attempt, not first-chunk.
+    ///
+    /// Attempt-scoped, so it excludes request parsing, guardrail scans,
+    /// routing, and the inter-attempt retry backoff. Summing a request's
+    /// attempts yields upstream time, NOT what the caller waited — that
+    /// is `downstream_latency_ms`.
+    pub upstream_latency_ms: u32,
+
+    /// Time to the upstream's first token, in milliseconds — measured
+    /// from the start of THIS attempt to the first upstream SSE chunk
+    /// carrying generated output (role-only preamble chunks don't
+    /// count). Same attempt scope as `upstream_latency_ms`, so the two
+    /// are directly comparable. 0 on non-streaming, error, and
     /// cache-hit paths (omitted from the wire via skip_serializing_if).
+    ///
+    /// This is what the UPSTREAM delivered on this attempt. What the
+    /// caller actually waited for is `downstream_latency_ms`, which also
+    /// covers gateway-side work — most visibly an output guardrail that
+    /// holds the stream back to mask it — and, when the request retried,
+    /// the earlier attempts too.
     #[serde(default, skip_serializing_if = "is_zero_u32")]
-    pub ttft_ms: u32,
+    pub upstream_ttft_ms: u32,
+
+    /// What the CALLER waited for, in milliseconds: from the gateway
+    /// receiving the request to it handing the client the first thing
+    /// it can use —
+    ///
+    /// - non-streaming: the complete response is written;
+    /// - streaming: the first token is forwarded downstream.
+    ///
+    /// Request-scoped (unlike the two `upstream_*` fields above), so it
+    /// spans request parsing, guardrail scans, every failed attempt,
+    /// the retry backoff, and any output-guardrail hold-back. Recorded
+    /// once per request, on the attempt that produced the terminal
+    /// response — including a failing one, so a request that never
+    /// succeeded still shows what its caller waited for.
+    ///
+    /// `downstream_latency_ms - upstream_ttft_ms` is the wait the final
+    /// attempt's upstream did NOT account for. On a first-try request
+    /// that is gateway-side work (parsing, guardrail scans, hold-back).
+    /// On a request that retried or failed over it also contains every
+    /// earlier attempt plus the backoff, so it is NOT gateway overhead
+    /// there — read it together with `attempt_index` before attributing
+    /// the difference to anything.
+    ///
+    /// Absent (0) on the non-terminal attempts of a request, and on any
+    /// path that never reached response delivery.
+    #[serde(default, skip_serializing_if = "is_zero_u32")]
+    pub downstream_latency_ms: u32,
 
     /// HTTP status code the proxy returned to the downstream caller.
     pub status_code: u16,
@@ -253,21 +296,21 @@ pub struct UsageEvent {
     // Each UsageEvent now represents ONE upstream attempt. A request
     // that fails over emits multiple events sharing `request_id` (the
     // grouping/trace key); they are ordered by `attempt_index`. This
-    // mirrors a per-call logging model — `status_code`, `latency_ms`,
-    // and `ttft_ms` are scoped to THIS attempt. Direct (non-routing)
-    // requests emit a single event with attempt_index=0,
-    // attempt_kind="initial".
+    // mirrors a per-call logging model — `status_code`,
+    // `upstream_latency_ms` and `upstream_ttft_ms` are scoped to THIS
+    // attempt. Direct (non-routing) requests emit a single event with
+    // attempt_index=0, attempt_kind="initial".
+    //
+    // The two latency families answer different questions and are
+    // deliberately measured against different clocks:
+    //
+    //   upstream_*   — attempt-scoped. How the upstream behaved on this
+    //                  one call. Comparable across attempts.
+    //   downstream_* — request-scoped, written once per request. What
+    //                  the caller waited for, gateway overhead included.
     //
-    // `latency_ms` measures that attempt alone: from the moment the
-    // attempt begins to the moment it settles — for a streamed attempt
-    // that is end-of-stream, not first-chunk (`ttft_ms` carries the
-    // first-token figure). It therefore excludes everything outside the
-    // attempt: request parsing, guardrail scans, routing, and the
-    // inter-attempt retry backoff. Summing a request's attempts yields
-    // upstream time, NOT the user-perceived total — that lives in the
-    // access log's `latency_ms`, which spans the whole request (and, on
-    // a streamed request, stops when the response head is handed to the
-    // client rather than when the body finishes).
+    // So a request's caller-facing latency is read off the single event
+    // carrying `downstream_latency_ms` — never by summing attempts.
     /// 0-based index of this attempt within the request. Together with
     /// `request_id` it uniquely identifies one attempt.
     #[serde(default)]
@@ -795,7 +838,7 @@ mod tests {
             requested_model: "smart-group".into(),
             prompt_tokens: 12,
             completion_tokens: 34,
-            latency_ms: 56,
+            upstream_latency_ms: 56,
             status_code: 200,
             cost_usd: 0.0012,
             guardrail_blocked: false,
@@ -831,7 +874,7 @@ mod tests {
         assert!(!json.contains("provider_request_id"));
         assert!(!json.contains("provider_model_version"));
         assert!(!json.contains("finish_reason"));
-        assert!(!json.contains("ttft_ms"));
+        assert!(!json.contains("upstream_ttft_ms"));
         // ProviderKey telemetry tag wire-compat (#302 M17 /
         // AISIX-Cloud#436). Pre-attribution DP images would emit
         // empty / false defaults, which must NOT appear on the wire.
@@ -953,7 +996,7 @@ mod tests {
             provider_request_id: "chatcmpl-abc".into(),
             provider_model_version: "gpt-4o-2024-08-06".into(),
             finish_reason: "stop".into(),
-            ttft_ms: 123,
+            upstream_ttft_ms: 123,
             ..Default::default()
         };
         let json = serde_json::to_string(&ev).unwrap();
@@ -964,7 +1007,7 @@ mod tests {
         assert!(json.contains(r#""provider_request_id":"chatcmpl-abc""#));
         assert!(json.contains(r#""provider_model_version":"gpt-4o-2024-08-06""#));
         assert!(json.contains(r#""finish_reason":"stop""#));
-        assert!(json.contains(r#""ttft_ms":123"#));
+        assert!(json.contains(r#""upstream_ttft_ms":123"#));
     }
 
     #[test]
@@ -978,7 +1021,7 @@ mod tests {
             status_code: 502,
             error_class: "upstream_status".into(),
             error_message: "upstream returned 502".into(),
-            latency_ms: 2000,
+            upstream_latency_ms: 2000,
             ..Default::default()
         };
         let json = serde_json::to_string(&failed).unwrap();

crates/aisix-proxy/src/a2a.rs

@@ -294,7 +294,10 @@ fn emit_a2a_usage(
         occurred_at: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
         api_key_id: auth.entry.id.clone(),
         status_code,
-        latency_ms: latency.as_millis().min(u32::MAX as u128) as u32,
+        // Single-attempt endpoint: the attempt spans the whole request, so
+        // the upstream figure and what the caller waited for coincide.
+        upstream_latency_ms: latency.as_millis().min(u32::MAX as u128) as u32,
+        downstream_latency_ms: latency.as_millis().min(u32::MAX as u128) as u32,
         inbound_protocol: "a2a".to_string(),
         a2a_agent_name: agent.to_string(),
         a2a_method: method.to_string(),

crates/aisix-proxy/src/audio.rs

@@ -1207,7 +1207,10 @@ fn emit_usage_event(
         requested_model: requested_model.to_string(),
         prompt_tokens,
         completion_tokens,
-        latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        // Single-attempt endpoint: the attempt spans the whole request, so
+        // the upstream figure and what the caller waited for coincide.
+        upstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        downstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
         status_code,
         inbound_protocol: "openai".to_string(),
         applied_guardrails: applied_guardrails.to_vec(),