From 3e7344787e6cbff3ece28ad016eb46cb21fcb8ed Mon Sep 17 00:00:00 2001 From: rock Date: Sun, 13 Sep 2026 22:02:56 +0900 Subject: [PATCH] feat: user identity + error name tracking in metrics - 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 --- crates/mem-cli/src/handlers/middleware.rs | 43 ++++++++++++++++++ crates/mem-cli/src/handlers/unified_query.rs | 9 +++- crates/mem-cli/src/http_server.rs | 12 +++++ crates/mem-cli/src/metrics.rs | 48 ++++++++++++++------ 4 files changed, 96 insertions(+), 16 deletions(-) diff --git a/crates/mem-cli/src/handlers/middleware.rs b/crates/mem-cli/src/handlers/middleware.rs index 7103ecb..210c8e6 100644 --- a/crates/mem-cli/src/handlers/middleware.rs +++ b/crates/mem-cli/src/handlers/middleware.rs @@ -61,6 +61,49 @@ pub fn validate_and_rate_limit( 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::(&payload_bytes) { + if let Some(sub) = payload.get("sub").and_then(|s| s.as_str()) { + return sub.to_string(); + } + } + } + + "anonymous".to_string() +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/mem-cli/src/handlers/unified_query.rs b/crates/mem-cli/src/handlers/unified_query.rs index d84a5c9..b42e9f0 100644 --- a/crates/mem-cli/src/handlers/unified_query.rs +++ b/crates/mem-cli/src/handlers/unified_query.rs @@ -130,13 +130,19 @@ pub async fn unified_query_handler( ) { QUERY_AUTH_FAILURES.inc(); QUERY_ERRORS_TOTAL.inc(); + ERRORS_BY_USER.inc(&["unknown", "/memory/query", "auth_failure"]); QUERY_IN_FLIGHT.dec(); 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 if let Err(response) = validate_unified_request(&body) { QUERY_ERRORS_TOTAL.inc(); + ERRORS_BY_USER.inc(&[&user_id, "/memory/query", "bad_request"]); QUERY_IN_FLIGHT.dec(); return response; } @@ -154,8 +160,9 @@ pub async fn unified_query_handler( Err(e) => { QUERY_EMBEDDING_FAILURES.inc(); QUERY_ERRORS_TOTAL.inc(); + ERRORS_BY_USER.inc(&[&user_id, "/memory/query", "embedding_failure"]); QUERY_IN_FLIGHT.dec(); - error!("Embedding failed: {}", e); + error!("Embedding failed for user={}: {}", user_id, e); return crate::handlers::response_builder::internal_error( "Failed to embed query" ); diff --git a/crates/mem-cli/src/http_server.rs b/crates/mem-cli/src/http_server.rs index dc3e743..173abbc 100644 --- a/crates/mem-cli/src/http_server.rs +++ b/crates/mem-cli/src/http_server.rs @@ -491,13 +491,19 @@ pub async fn ingest_handler( Err(e) => { INGEST_AUTH_FAILURES.inc(); INGEST_ERRORS_TOTAL.inc(); + ERRORS_BY_USER.inc(&["unknown", "/memory/ingest", "auth_failure"]); INGEST_IN_FLIGHT.dec(); return e; } }; + + let user_id = &claims.sub; + REQUESTS_BY_USER.inc(&[user_id, "/memory/ingest"]); + if !has_capability(&claims, "memory:write") { INGEST_AUTH_FAILURES.inc(); INGEST_ERRORS_TOTAL.inc(); + ERRORS_BY_USER.inc(&[user_id, "/memory/ingest", "forbidden"]); INGEST_IN_FLIGHT.dec(); return HttpResponse::Forbidden().json(json!({ "error": "forbidden", @@ -506,6 +512,7 @@ pub async fn ingest_handler( } if let Err(e) = check_rate_limit(&claims, &state, "/memory/ingest") { INGEST_RATE_LIMITED.inc(); + ERRORS_BY_USER.inc(&[user_id, "/memory/ingest", "rate_limited"]); INGEST_IN_FLIGHT.dec(); return e; } @@ -1044,12 +1051,17 @@ pub async fn context_handler( Ok(c) => c, Err(e) => { CONTEXT_ERRORS_TOTAL.inc(); + ERRORS_BY_USER.inc(&["unknown", "/memory/context", "auth_failure"]); return e; } }; + let user_id = &claims.sub; + REQUESTS_BY_USER.inc(&[user_id, "/memory/context"]); + if !has_capability(&claims, "memory:read") { CONTEXT_ERRORS_TOTAL.inc(); + ERRORS_BY_USER.inc(&[user_id, "/memory/context", "forbidden"]); return HttpResponse::Forbidden().json(json!({ "error": "forbidden", "reason": "missing capability: memory:read" diff --git a/crates/mem-cli/src/metrics.rs b/crates/mem-cli/src/metrics.rs index 5d3dbbe..bf60ff0 100644 --- a/crates/mem-cli/src/metrics.rs +++ b/crates/mem-cli/src/metrics.rs @@ -334,6 +334,20 @@ pub static REQUEST_ERRORS_BY_STATUS: Lazy = Lazy::new(|| "memory_request_errors_by_status", "Request errors by HTTP status code", &["status", "endpoint"])); +/// Errors with user identity and error name +/// Labels: [user_id, endpoint, error_name] +pub static ERRORS_BY_USER: Lazy = 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 = Lazy::new(|| + LabeledCounter::new( + "memory_requests_by_user", "Requests by user identity", + &["user_id", "endpoint"])); + // ═══════════════════════════════════════════════════════════ // O8: Ingest rate pattern tracking (IR1-IR10) // ═══════════════════════════════════════════════════════════ @@ -525,25 +539,29 @@ pub fn render_metrics() -> String { gauge!(DB_TABLE_CHUNK_ROWS); // Labeled counter: errors by status - { - let map = REQUEST_ERRORS_BY_STATUS.values.lock().unwrap(); - if !map.is_empty() { - out.push_str(&format!("# HELP {} {}\n# TYPE {} counter\n", - REQUEST_ERRORS_BY_STATUS.name, REQUEST_ERRORS_BY_STATUS.help, - 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)); - } - } - } - } + render_labeled_counter(&mut out, &REQUEST_ERRORS_BY_STATUS); + // Labeled counter: errors by user + render_labeled_counter(&mut out, &ERRORS_BY_USER); + // Labeled counter: requests by user + render_labeled_counter(&mut out, &REQUESTS_BY_USER); 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 = 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 pub async fn metrics_handler() -> actix_web::HttpResponse { actix_web::HttpResponse::Ok()