feat: user identity + error name tracking in metrics
CI / CI (pull_request) Successful in 12m36s

- ERRORS_BY_USER: labeled counter {user_id, endpoint, error_name}
- REQUESTS_BY_USER: labeled counter {user_id, endpoint}
- extract_user_id(): decode JWT sub claim from Authorization header
- Error names: auth_failure, forbidden, rate_limited, bad_request, embedding_failure
- Ingest handler: tracks user_id from claims.sub
- Query handler: tracks user_id from JWT decode
- Context handler: tracks user_id from claims.sub
- render_labeled_counter(): generic Prometheus label renderer
- User identity from gateway JWT (claims.sub per API.md)
- 515 tests passing
This commit is contained in:
2026-09-13 22:02:56 +09:00
parent 49dcf2616c
commit 3e7344787e
4 changed files with 96 additions and 16 deletions
+43
View File
@@ -61,6 +61,49 @@ pub fn validate_and_rate_limit(
Ok(()) Ok(())
} }
/// Extract user identity from JWT claims (sub field)
///
/// Tries to decode JWT from Authorization header to get `sub` claim.
/// Falls back to "anonymous" if auth is disabled or header missing.
/// Used by metrics to track errors/requests per user.
pub fn extract_user_id(req: &HttpRequest, state: &AppState) -> String {
// If auth disabled, check synthetic claims
if state.jwt_validator.is_none() {
return "anonymous".to_string();
}
// Try to extract sub from JWT
let token = req.headers()
.get("Authorization")
.and_then(|h| h.to_str().ok())
.and_then(|h| h.strip_prefix("Bearer "))
.unwrap_or("");
if token.is_empty() {
return "anonymous".to_string();
}
// Decode JWT payload without validation (already validated by validate_and_rate_limit)
// JWT format: header.payload.signature
let parts: Vec<&str> = token.split('.').collect();
if parts.len() != 3 {
return "anonymous".to_string();
}
// Decode base64 payload
use base64::Engine;
let engine = base64::engine::general_purpose::URL_SAFE_NO_PAD;
if let Ok(payload_bytes) = engine.decode(parts[1]) {
if let Ok(payload) = serde_json::from_slice::<serde_json::Value>(&payload_bytes) {
if let Some(sub) = payload.get("sub").and_then(|s| s.as_str()) {
return sub.to_string();
}
}
}
"anonymous".to_string()
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
+8 -1
View File
@@ -130,13 +130,19 @@ pub async fn unified_query_handler(
) { ) {
QUERY_AUTH_FAILURES.inc(); QUERY_AUTH_FAILURES.inc();
QUERY_ERRORS_TOTAL.inc(); QUERY_ERRORS_TOTAL.inc();
ERRORS_BY_USER.inc(&["unknown", "/memory/query", "auth_failure"]);
QUERY_IN_FLIGHT.dec(); QUERY_IN_FLIGHT.dec();
return response; return response;
} }
// Extract user_id from JWT sub claim via state (set by validate_and_rate_limit)
let user_id = crate::handlers::middleware::extract_user_id(&req, &state);
REQUESTS_BY_USER.inc(&[&user_id, "/memory/query"]);
// 2. Validate input // 2. Validate input
if let Err(response) = validate_unified_request(&body) { if let Err(response) = validate_unified_request(&body) {
QUERY_ERRORS_TOTAL.inc(); QUERY_ERRORS_TOTAL.inc();
ERRORS_BY_USER.inc(&[&user_id, "/memory/query", "bad_request"]);
QUERY_IN_FLIGHT.dec(); QUERY_IN_FLIGHT.dec();
return response; return response;
} }
@@ -154,8 +160,9 @@ pub async fn unified_query_handler(
Err(e) => { Err(e) => {
QUERY_EMBEDDING_FAILURES.inc(); QUERY_EMBEDDING_FAILURES.inc();
QUERY_ERRORS_TOTAL.inc(); QUERY_ERRORS_TOTAL.inc();
ERRORS_BY_USER.inc(&[&user_id, "/memory/query", "embedding_failure"]);
QUERY_IN_FLIGHT.dec(); QUERY_IN_FLIGHT.dec();
error!("Embedding failed: {}", e); error!("Embedding failed for user={}: {}", user_id, e);
return crate::handlers::response_builder::internal_error( return crate::handlers::response_builder::internal_error(
"Failed to embed query" "Failed to embed query"
); );
+12
View File
@@ -491,13 +491,19 @@ pub async fn ingest_handler(
Err(e) => { Err(e) => {
INGEST_AUTH_FAILURES.inc(); INGEST_AUTH_FAILURES.inc();
INGEST_ERRORS_TOTAL.inc(); INGEST_ERRORS_TOTAL.inc();
ERRORS_BY_USER.inc(&["unknown", "/memory/ingest", "auth_failure"]);
INGEST_IN_FLIGHT.dec(); INGEST_IN_FLIGHT.dec();
return e; return e;
} }
}; };
let user_id = &claims.sub;
REQUESTS_BY_USER.inc(&[user_id, "/memory/ingest"]);
if !has_capability(&claims, "memory:write") { if !has_capability(&claims, "memory:write") {
INGEST_AUTH_FAILURES.inc(); INGEST_AUTH_FAILURES.inc();
INGEST_ERRORS_TOTAL.inc(); INGEST_ERRORS_TOTAL.inc();
ERRORS_BY_USER.inc(&[user_id, "/memory/ingest", "forbidden"]);
INGEST_IN_FLIGHT.dec(); INGEST_IN_FLIGHT.dec();
return HttpResponse::Forbidden().json(json!({ return HttpResponse::Forbidden().json(json!({
"error": "forbidden", "error": "forbidden",
@@ -506,6 +512,7 @@ pub async fn ingest_handler(
} }
if let Err(e) = check_rate_limit(&claims, &state, "/memory/ingest") { if let Err(e) = check_rate_limit(&claims, &state, "/memory/ingest") {
INGEST_RATE_LIMITED.inc(); INGEST_RATE_LIMITED.inc();
ERRORS_BY_USER.inc(&[user_id, "/memory/ingest", "rate_limited"]);
INGEST_IN_FLIGHT.dec(); INGEST_IN_FLIGHT.dec();
return e; return e;
} }
@@ -1044,12 +1051,17 @@ pub async fn context_handler(
Ok(c) => c, Ok(c) => c,
Err(e) => { Err(e) => {
CONTEXT_ERRORS_TOTAL.inc(); CONTEXT_ERRORS_TOTAL.inc();
ERRORS_BY_USER.inc(&["unknown", "/memory/context", "auth_failure"]);
return e; return e;
} }
}; };
let user_id = &claims.sub;
REQUESTS_BY_USER.inc(&[user_id, "/memory/context"]);
if !has_capability(&claims, "memory:read") { if !has_capability(&claims, "memory:read") {
CONTEXT_ERRORS_TOTAL.inc(); CONTEXT_ERRORS_TOTAL.inc();
ERRORS_BY_USER.inc(&[user_id, "/memory/context", "forbidden"]);
return HttpResponse::Forbidden().json(json!({ return HttpResponse::Forbidden().json(json!({
"error": "forbidden", "error": "forbidden",
"reason": "missing capability: memory:read" "reason": "missing capability: memory:read"
+33 -15
View File
@@ -334,6 +334,20 @@ pub static REQUEST_ERRORS_BY_STATUS: Lazy<LabeledCounter> = Lazy::new(||
"memory_request_errors_by_status", "Request errors by HTTP status code", "memory_request_errors_by_status", "Request errors by HTTP status code",
&["status", "endpoint"])); &["status", "endpoint"]));
/// Errors with user identity and error name
/// Labels: [user_id, endpoint, error_name]
pub static ERRORS_BY_USER: Lazy<LabeledCounter> = Lazy::new(||
LabeledCounter::new(
"memory_errors_by_user", "Errors by user identity and error type",
&["user_id", "endpoint", "error_name"]));
/// Requests by user identity
/// Labels: [user_id, endpoint]
pub static REQUESTS_BY_USER: Lazy<LabeledCounter> = Lazy::new(||
LabeledCounter::new(
"memory_requests_by_user", "Requests by user identity",
&["user_id", "endpoint"]));
// ═══════════════════════════════════════════════════════════ // ═══════════════════════════════════════════════════════════
// O8: Ingest rate pattern tracking (IR1-IR10) // O8: Ingest rate pattern tracking (IR1-IR10)
// ═══════════════════════════════════════════════════════════ // ═══════════════════════════════════════════════════════════
@@ -525,25 +539,29 @@ pub fn render_metrics() -> String {
gauge!(DB_TABLE_CHUNK_ROWS); gauge!(DB_TABLE_CHUNK_ROWS);
// Labeled counter: errors by status // Labeled counter: errors by status
{ render_labeled_counter(&mut out, &REQUEST_ERRORS_BY_STATUS);
let map = REQUEST_ERRORS_BY_STATUS.values.lock().unwrap(); // Labeled counter: errors by user
if !map.is_empty() { render_labeled_counter(&mut out, &ERRORS_BY_USER);
out.push_str(&format!("# HELP {} {}\n# TYPE {} counter\n", // Labeled counter: requests by user
REQUEST_ERRORS_BY_STATUS.name, REQUEST_ERRORS_BY_STATUS.help, render_labeled_counter(&mut out, &REQUESTS_BY_USER);
REQUEST_ERRORS_BY_STATUS.name));
for (key, val) in map.iter() {
let parts: Vec<&str> = key.split(',').collect();
if parts.len() == 2 {
out.push_str(&format!("{}{{status=\"{}\",endpoint=\"{}\"}} {}\n",
REQUEST_ERRORS_BY_STATUS.name, parts[0], parts[1], val));
}
}
}
}
out out
} }
/// Render a labeled counter in Prometheus format
fn render_labeled_counter(out: &mut String, lc: &LabeledCounter) {
let map = lc.values.lock().unwrap();
if map.is_empty() { return; }
out.push_str(&format!("# HELP {} {}\n# TYPE {} counter\n", lc.name, lc.help, lc.name));
for (key, val) in map.iter() {
let parts: Vec<&str> = key.split(',').collect();
let labels: Vec<String> = lc.label_names.iter().zip(parts.iter())
.map(|(name, val)| format!("{}=\"{}\"", name, val))
.collect();
out.push_str(&format!("{}{{{}}} {}\n", lc.name, labels.join(","), val));
}
}
/// GET /metrics handler /// GET /metrics handler
pub async fn metrics_handler() -> actix_web::HttpResponse { pub async fn metrics_handler() -> actix_web::HttpResponse {
actix_web::HttpResponse::Ok() actix_web::HttpResponse::Ok()