feat(telemetry): attribute usage events and their counters to the org member

crates/aisix-obs/src/metrics.rs

@@ -259,8 +259,18 @@ pub const M_GUARDRAIL_LATENCY_SECONDS: &str = "aisix_guardrail_latency_seconds";
 /// - `handler`: which OpenAI-shape handler emitted (chat /
 ///   embeddings / responses / completions / rerank / audio /
 ///   images / messages). Fixed enumeration, low cardinality.
-/// - `status_code`: bucketed as `2xx` / `4xx` / `5xx` (avoid the
-///   1000-value cardinality blowup of raw u16 codes).
+/// - `status_code`: bucketed as `2xx` / `4xx` / `5xx`.
+/// - `status`: the raw code (`429`, `502`, ...), so a query can name one
+///   failure mode instead of a whole family (AISIX-Cloud#1389). Named to
+///   match `aisix_proxy_requests_total`'s own raw-code label, so one
+///   spelling works across both families — and deliberately NOT
+///   `http_status_code`, which contains `status_code` as a substring and
+///   would make every `status_code="…"` matcher ambiguous. It adds no
+///   series over `status_code` alone (the raw code determines the family),
+///   and `status_code` is kept so dashboards and alerts written against
+///   `status_code="4xx"` keep working unchanged.
+/// - `user_id`: the org member behind the request, from
+///   [`UsageEventLabels`].
 /// - `inbound_protocol`: `openai` / `anthropic`. Matches the
 ///   wire-level field on UsageEvent.
 /// - `upstream_protocol`: the wire protocol of the ProviderKey the event
@@ -2013,8 +2023,11 @@ impl Metrics {
     /// that invariant exists only on dimensions BOTH sides carry. The
     /// protocol is a function of the ProviderKey row `provider_key_id`
     /// already names, so like `provider_key_name` it adds no series here.
-    /// (`handler` / `status_code` / `inbound_protocol` are the emit
-    /// counter's own arguments, not attribution, and stay one-sided.)
+    /// (`handler` / `status_code` / `status` / `inbound_protocol`
+    /// are the emit counter's own arguments, not attribution, and stay
+    /// one-sided — `emitted == delivered + dropped` is an invariant over
+    /// the attribution dimensions, which is why `user_id` DOES appear on
+    /// both.)
     pub fn record_usage_event_drop(&self, reason: &str, labels: UsageEventLabels<'_>) {
         self.cached_counter(
             M_USAGE_EVENT_DROPS_TOTAL,
@@ -2024,6 +2037,7 @@ impl Metrics {
                 k.label(labels.model);
                 k.label(labels.provider_key_id);
                 k.label(labels.provider_key_name);
+                k.label(labels.user_id);
                 k.label(labels.upstream_protocol);
             },
             || {
@@ -2033,6 +2047,7 @@ impl Metrics {
                     "model" => labels.model.to_string(),
                     "provider_key_id" => labels.provider_key_id.to_string(),
                     "provider_key_name" => labels.provider_key_name.to_string(),
+                    "user_id" => labels.user_id.to_string(),
                     "upstream_protocol" => labels.upstream_protocol.to_string(),
                 )
             },
@@ -2074,22 +2089,26 @@ impl Metrics {
             |k| {
                 k.label(handler);
                 k.label(status_class);
+                k.label_u16(status_code);
                 k.label(inbound_protocol);
                 k.label(labels.model);
                 k.label(labels.provider_key_id);
                 k.label(labels.provider_key_name);
+                k.label(labels.user_id);
                 k.label(labels.upstream_protocol);
             },
             || {
                 metrics::counter!(
                     M_USAGE_EVENT_EMITS_TOTAL,
                     "handler" => handler,
                     "status_code" => status_class,
+                    "status" => status_code.to_string(),
                     "inbound_protocol" => inbound_protocol,
                     "upstream_protocol" => labels.upstream_protocol.to_string(),
                     "model" => labels.model.to_string(),
                     "provider_key_id" => labels.provider_key_id.to_string(),
                     "provider_key_name" => labels.provider_key_name.to_string(),
+                    "user_id" => labels.user_id.to_string(),
                 )
             },
         );
@@ -2683,6 +2702,19 @@ pub struct UsageEventLabels<'a> {
     pub model: &'a str,
     pub provider_key_id: &'a str,
     pub provider_key_name: &'a str,
+    /// Org member the authenticating key belongs to (AISIX-Cloud#1389) —
+    /// the same `user_id` `aisix_proxy_requests_total` carries, so a
+    /// member's request rate and their usage-event rate slice alike, and
+    /// the readable `user_name` is joinable from that family rather than
+    /// duplicated here. `unknown` when the key is bound to no member, or
+    /// when auth never resolved a key.
+    ///
+    /// This is attribution, so it sits on BOTH counters: "which member's
+    /// usage records were lost" is exactly the question the shared label
+    /// set exists to keep answerable. `UsageSink::try_emit` fills it from
+    /// the event's own `user_id`, so the counter and the row cp-api
+    /// persists can never name different people.
+    pub user_id: &'a str,
     /// The upstream wire protocol of the key named by `provider_key_id`
     /// (AISIX-Cloud#1403) — the same value, resolved the same way, that
     /// the request and usage families carry, so an operator can align
@@ -2703,6 +2735,7 @@ impl Default for UsageEventLabels<'_> {
             model: "unknown",
             provider_key_id: "unknown",
             provider_key_name: "unknown",
+            user_id: "unknown",
             upstream_protocol: "unknown",
         }
     }
@@ -4281,6 +4314,7 @@ mod tests {
         model: "gpt-4o",
         provider_key_id: "pk-1",
         provider_key_name: "openai-prod",
+        user_id: "member-1",
         upstream_protocol: "openai",
     };
 
@@ -4349,14 +4383,14 @@ mod tests {
 
         assert!(
             rendered.contains(
-                "handler=\"messages\",status_code=\"2xx\",\
+                "handler=\"messages\",status_code=\"2xx\",status=\"200\",\
                  inbound_protocol=\"anthropic\",upstream_protocol=\"openai\""
             ),
             "cross-protocol sample must report the upstream's protocol:\n{rendered}"
         );
         assert!(
             rendered.contains(
-                "handler=\"chat\",status_code=\"4xx\",\
+                "handler=\"chat\",status_code=\"4xx\",status=\"401\",\
                  inbound_protocol=\"openai\",upstream_protocol=\"unknown\""
             ),
             "an unresolved upstream must read `unknown`, never borrow the \
@@ -4414,7 +4448,7 @@ mod tests {
         };
         // The emit counter's own arguments — a surface of the emitting
         // handler, not of the event's attribution.
-        let emit_only: BTreeSet<String> = ["handler", "status_code", "inbound_protocol"]
+        let emit_only: BTreeSet<String> = ["handler", "status_code", "status", "inbound_protocol"]
             .into_iter()
             .map(str::to_string)
             .collect();
@@ -4423,7 +4457,9 @@ mod tests {
             .cloned()
             .collect();
         assert!(
-            attribution.contains("model") && attribution.contains("provider_key_id"),
+            attribution.contains("model")
+                && attribution.contains("provider_key_id")
+                && attribution.contains("user_id"),
             "the attribution set looks wrong: {attribution:?}"
         );
 

crates/aisix-obs/src/otlp_http_sink.rs

@@ -822,6 +822,12 @@ fn event_attributes(record: &SinkRecord, exporter_name: &str) -> Vec<Value> {
         // spans back to the AISIX api_key dashboard.
         attributes.push(attr_string("aisix.api_key_id", &event.api_key_id));
     }
+    // The org member behind the credential (AISIX-Cloud#1389). Exported
+    // beside the key rather than derived from it downstream: a member can
+    // hold several keys, and a key can be rebound to someone else.
+    if !event.user_id.is_empty() {
+        attributes.push(attr_string("aisix.user_id", &event.user_id));
+    }
     if !event.model_id.is_empty() {
         attributes.push(attr_string("aisix.model_id", &event.model_id));
     }

crates/aisix-obs/src/usage.rs

@@ -76,6 +76,17 @@ pub struct UsageEvent {
     #[serde(default)]
     pub api_key_id: String,
 
+    /// UUID of the org member the authenticating ApiKey is owned by
+    /// (`ApiKey.user_id`), snapshotted at request time so the Logs
+    /// member filter keeps naming who actually made the call
+    /// (AISIX-Cloud#1389). Resolving it from `api_key_id` at query time
+    /// instead would re-attribute a key's whole history the moment an
+    /// operator rebinds it, and lose the attribution entirely once the
+    /// key is deleted. Empty when the key is bound to no member, or
+    /// when auth failed before resolution; cp-api stores empty as NULL.
+    #[serde(default, skip_serializing_if = "String::is_empty")]
+    pub user_id: String,
+
     /// The model alias exactly as the client sent it in the request
     /// body (`model` field) — a Model-Group name for routed requests,
     /// a direct model's display name otherwise. `model_id` records the
@@ -775,8 +786,25 @@ impl UsageSink {
     /// records did we lose" answerable. The event itself cannot supply
     /// them: its `requested_model` is caller-controlled text (#451) and it
     /// carries no ProviderKey id at all.
+    ///
+    /// The one attribution dimension that DOES come off the event is
+    /// `user_id` (AISIX-Cloud#1389): it is a resolved ApiKey field, not
+    /// caller text, and taking it here rather than from each handler's
+    /// label builder is what makes the counter and the row cp-api persists
+    /// structurally incapable of naming different members.
     pub fn try_emit(&self, handler: &'static str, event: UsageEvent, labels: UsageEventLabels<'_>) {
         log_provider_call(handler, &event);
+        // Owned because `event` is moved into the channel below while the
+        // drop counter still needs the label.
+        let user_id = event.user_id.clone();
+        let labels = UsageEventLabels {
+            user_id: if user_id.is_empty() {
+                "unknown"
+            } else {
+                user_id.as_str()
+            },
+            ..labels
+        };
         // Normalise inbound_protocol to a fixed `&'static str` set at
         // the boundary (audit MEDIUM-3). This both kills the heap
         // alloc per call AND pins prometheus cardinality at the type
@@ -1175,14 +1203,18 @@ mod tests {
         sink.try_emit(
             "chat",
             UsageEvent {
-                status_code: 200,
+                status_code: 429,
                 inbound_protocol: "openai".into(),
+                // Attribution the sink reads off the event itself, not off
+                // the label set the handler built.
+                user_id: "member-1".into(),
                 ..Default::default()
             },
             UsageEventLabels {
                 model: "customer-chat",
                 provider_key_id: "pk-1",
                 provider_key_name: "openai-prod",
+                user_id: "unknown",
                 upstream_protocol: "openai",
             },
         );
@@ -1196,6 +1228,11 @@ mod tests {
                 ("model", "customer-chat"),
                 ("provider_key_id", "pk-1"),
                 ("provider_key_name", "openai-prod"),
+                ("user_id", "member-1"),
+                // The raw code sits beside the family, so a query can name
+                // one failure mode without giving up the family rollup.
+                ("status_code", "4xx"),
+                ("status", "429"),
             ],
         );
         let dropped = parse_counter_value(
@@ -1206,6 +1243,7 @@ mod tests {
                 ("model", "customer-chat"),
                 ("provider_key_id", "pk-1"),
                 ("provider_key_name", "openai-prod"),
+                ("user_id", "member-1"),
             ],
         );
         assert_eq!(

crates/aisix-proxy/src/a2a.rs

@@ -889,7 +889,11 @@ fn emit_a2a_usage(
         guardrail_blocked,
         ..Default::default()
     };
-    crate::usage_attr::apply_jwt_identity(&mut event, auth.jwt.as_ref());
+    crate::usage_attr::apply_caller_identity(
+        &mut event,
+        auth.jwt.as_ref(),
+        auth.key().user_id.as_deref(),
+    );
     // The client-perceived duration of the call. Nothing else records it for
     // `/a2a`: the handler returns the moment a stream's response head is out,
     // so `aisix_proxy_request_duration_seconds` times only how long a stream

crates/aisix-proxy/src/audio.rs

@@ -2104,7 +2104,11 @@ fn emit_usage_event(
     // responses (AISIX-Cloud#867 parity).
     crate::usage_attr::apply_pk_telemetry(&mut event, pk);
     // Handler label "audio" — bucketed prometheus counter (#408).
-    crate::usage_attr::apply_jwt_identity(&mut event, client.jwt.as_ref());
+    crate::usage_attr::apply_caller_identity(
+        &mut event,
+        client.jwt.as_ref(),
+        client.caller.user_id.as_deref(),
+    );
     let usage_model =
         crate::usage_attr::usage_event_model_label(snap, &event.requested_model).into_owned();
     crate::usage_attr::emit_usage(

crates/aisix-proxy/src/chat.rs

@@ -4409,7 +4409,11 @@ fn emit_usage_event(
         // MCP attribution does not apply to the chat path.
         ..Default::default()
     };
-    crate::usage_attr::apply_jwt_identity(&mut event, client.jwt.as_ref());
+    crate::usage_attr::apply_caller_identity(
+        &mut event,
+        client.jwt.as_ref(),
+        client.caller.user_id.as_deref(),
+    );
     // Guardrail outcome counters (#379). Recorded here — the one place every
     // chat path (success / error / streaming / cache-hit) funnels through —
     // from the same guardrail fields the UsageEvent carries.

crates/aisix-proxy/src/completions.rs

@@ -766,7 +766,11 @@ fn emit_usage_event(
         ..Default::default()
     };
     crate::usage_attr::apply_pk_telemetry(&mut event, pk);
-    crate::usage_attr::apply_jwt_identity(&mut event, client.jwt.as_ref());
+    crate::usage_attr::apply_caller_identity(
+        &mut event,
+        client.jwt.as_ref(),
+        client.caller.user_id.as_deref(),
+    );
     let usage_model =
         crate::usage_attr::usage_event_model_label(snap, &event.requested_model).into_owned();
     crate::usage_attr::emit_usage(

crates/aisix-proxy/src/embeddings.rs

@@ -715,7 +715,11 @@ fn emit_usage_event(
     };
     crate::usage_attr::apply_pk_telemetry(&mut event, pk);
     // Handler label "embeddings" — bucketed prometheus counter (#408).
-    crate::usage_attr::apply_jwt_identity(&mut event, client.jwt.as_ref());
+    crate::usage_attr::apply_caller_identity(
+        &mut event,
+        client.jwt.as_ref(),
+        client.caller.user_id.as_deref(),
+    );
     let usage_model =
         crate::usage_attr::usage_event_model_label(snap, &event.requested_model).into_owned();
     crate::usage_attr::emit_usage(

crates/aisix-proxy/src/images.rs

@@ -551,7 +551,11 @@ pub(crate) fn emit_usage_event(
     };
     crate::usage_attr::apply_pk_telemetry(&mut event, pk);
     // Handler label "images" — bucketed prometheus counter (#408).
-    crate::usage_attr::apply_jwt_identity(&mut event, client.jwt.as_ref());
+    crate::usage_attr::apply_caller_identity(
+        &mut event,
+        client.jwt.as_ref(),
+        client.caller.user_id.as_deref(),
+    );
     let usage_model =
         crate::usage_attr::usage_event_model_label(snap, &event.requested_model).into_owned();
     crate::usage_attr::emit_usage(

crates/aisix-proxy/src/jobs.rs

@@ -623,7 +623,11 @@ fn emit_job_usage_event(
     };
     let pk = crate::usage_attr::ResolvedPk::resolve(snap, &target.pk_entry.id);
     crate::usage_attr::apply_pk_telemetry(&mut event, &pk);
-    crate::usage_attr::apply_jwt_identity(&mut event, auth.jwt.as_ref());
+    crate::usage_attr::apply_caller_identity(
+        &mut event,
+        auth.jwt.as_ref(),
+        auth.key().user_id.as_deref(),
+    );
     let usage_model =
         crate::usage_attr::usage_event_model_label(snap, &event.requested_model).into_owned();
     crate::usage_attr::emit_usage(
@@ -1802,6 +1806,7 @@ fn maybe_attribute_batch(
 
     let state = state.clone();
     let api_key_id = auth.entry.id.clone();
+    let user_id = auth.entry.value.user_id.clone();
     let jwt = auth.jwt.clone();
     let model_id = target.model_entry.id.clone();
     let display_name = target.display_name().to_string();
@@ -1818,6 +1823,7 @@ fn maybe_attribute_batch(
             &state,
             &api_key_id,
             jwt.as_ref(),
+            user_id.as_deref(),
             &model_id,
             &display_name,
             cost.as_ref(),
@@ -1850,6 +1856,7 @@ async fn attribute_batch_usage(
     state: &ProxyState,
     api_key_id: &str,
     jwt: Option<&std::sync::Arc<crate::auth::JwtIdentity>>,
+    user_id: Option<&str>,
     model_id: &str,
     display_name: &str,
     cost: Option<&aisix_core::models::model::ModelCost>,
@@ -1962,7 +1969,7 @@ async fn attribute_batch_usage(
         crate::usage_attr::apply_pk_telemetry(&mut event, &pk);
         // Attribution names the identity that observed completion — the
         // same caller the event's api_key_id already reflects.
-        crate::usage_attr::apply_jwt_identity(&mut event, jwt);
+        crate::usage_attr::apply_caller_identity(&mut event, jwt, user_id);
         let usage_model =
             crate::usage_attr::usage_event_model_label(&snap, &event.requested_model).into_owned();
         state.usage_sink.try_emit(

crates/aisix-proxy/src/mcp.rs

@@ -1188,7 +1188,11 @@ fn emit_tool_call_usage(
             .unwrap_or_default(),
         ..Default::default()
     };
-    crate::usage_attr::apply_jwt_identity(&mut event, auth.jwt.as_ref());
+    crate::usage_attr::apply_caller_identity(
+        &mut event,
+        auth.jwt.as_ref(),
+        auth.key().user_id.as_deref(),
+    );
     crate::usage_attr::apply_auth_type(&mut event, auth);
     // A tool call resolves neither a model nor a ProviderKey, so the
     // attribution labels are the placeholder — present so this family has

crates/aisix-proxy/src/messages.rs

@@ -2996,7 +2996,11 @@ fn emit_anthropic_usage_event(
     };
     // Handler label "messages" — Anthropic /v1/messages inbound
     // path. Bucketed prometheus counter (#408).
-    crate::usage_attr::apply_jwt_identity(&mut event, client.jwt.as_ref());
+    crate::usage_attr::apply_caller_identity(
+        &mut event,
+        client.jwt.as_ref(),
+        client.caller.user_id.as_deref(),
+    );
     let usage_model = crate::usage_attr::usage_event_model_label(snap, &event.requested_model);
     // The metric code below still reads `event`, so the chokepoint gets
     // its own copy (it stamps `trace_id` on the emitted one).