Files
poimen-memory/crates/mem-cli/src/handlers/middleware.rs
T
poimenandrock 4169effd8a
CI / CI (push) Successful in 12m36s
Deploy / Tag & Push Latest (push) Successful in 1m56s
feat: complete observability stack (O1-O13) (#52)
## Complete Observability Stack (O1-O13)

Implements all 13 observability issues in a single PR. 119 metrics total.

### Commits (one per issue)

| Issue | Title | Metrics |
|-------|-------|---------|
| **O10** | Prometheus metrics module + /metrics endpoint | Foundation |
| **O1** | Instrument ingest handler | I1-I12 (12) |
| **O2** | Instrument query handler | Q1-Q12 (12) |
| **O3** | Instrument context endpoint | C1-C8 (8) |
| **O4** | Relevance judge | R1-R9 (9) |
| **O5** | Write volume + storage metrics | W1-W12 (12) |
| **O6** | Pod resource observability | P1-P13 |
| **O7** | Availability + dependency health | A1-A10 (10) |
| **O8** | Ingest rate pattern tracking | IR1-IR10 (10) |
| **O9** | Postgres internal observability | PG1-PG33 |
| **O11** | Grafana dashboard | 12 panels |
| **O12** | Prometheus alerting rules | 11 alerts |
| **O13** | Relevance evaluation CronJob | K8s manifest |

### Key Changes

- **metrics.rs**: Zero-dependency Prometheus metrics (Counter, Gauge, Histogram, Timer)
- **GET /metrics**: Prometheus text exposition format endpoint
- **Ingest/Query/Context handlers**: Instrumented with latency, errors, auth failures
- **Health check**: DB dependency check with latency tracking
- **Background task**: Periodic DB stats collection (entity/edge counts, pool stats)
- **Relevance judge**: Threshold-based eval with precision/recall/F1 tracking
- **Grafana dashboard**: 12 panels covering all metric groups
- **Alert rules**: 11 PrometheusRule alerts (availability, latency, errors, quality)
- **CronJob**: Periodic relevance evaluation with sample queries

### Testing

- 506 tests passing (0 failures)
- All metrics modules have unit tests
- Relevance judge: 4 tests

### Deploy

```bash
# Grafana dashboard
kubectl apply -f k8s/infra/grafana-dashboard.json

# Prometheus alerts
kubectl apply -f k8s/infra/prometheus-alerts.yaml

# Relevance eval CronJob
kubectl apply -f k8s/infra/relevance-eval-cronjob.yaml
```

Closes #27 #28 #29 #30 #31 #32 #33 #34 #35 #36 #37 #38 #39

---------

Co-authored-by: rock <[email protected]>
Reviewed-on: #52
Co-authored-by: poimen <[email protected]>
2026-09-13 13:53:50 +00:00

125 lines
3.7 KiB
Rust

/// Handler middleware utilities
///
/// Centralized JWT validation + rate limiting for all HTTP handlers.
/// Eliminates boilerplate across endpoints, improves testability.
use actix_web::{HttpRequest, HttpResponse};
use serde_json::json;
use crate::http_server::AppState;
/// Result type for middleware operations
pub type MiddlewareResult<T> = Result<T, HttpResponse>;
/// Validate JWT token + check rate limit
///
/// Handles:
/// 1. Extract Authorization header
/// 2. Validate JWT (if auth enabled)
/// 3. Check rate limit (if limiter enabled)
/// 4. Return error response on failure
///
/// # Usage
/// ```ignore
/// validate_and_rate_limit(&req, &state, "compact", 10)?;
/// // If we get here, both JWT and rate limit checks passed
/// ```
pub fn validate_and_rate_limit(
req: &HttpRequest,
state: &AppState,
endpoint: &str,
rate_limit: u32,
) -> MiddlewareResult<()> {
// 1. JWT validation (if enabled)
if let Some(jwt_validator) = &state.jwt_validator {
let auth_header = req
.headers()
.get("Authorization")
.and_then(|h| h.to_str().ok())
.ok_or_else(|| {
HttpResponse::Unauthorized().json(json!({
"error": "Missing Authorization header"
}))
})?;
crate::jwt_validator::JwtValidator::extract_bearer_token(auth_header).map_err(|e| {
HttpResponse::Unauthorized().json(json!({
"error": format!("JWT validation failed: {}", e)
}))
})?;
}
// 2. Rate limiting (if enabled)
state
.rate_limiter
.check("default", endpoint)
.map_err(|e| {
HttpResponse::TooManyRequests().json(json!({
"error": format!("Rate limit exceeded: {}", e.reason())
}))
})?;
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)]
mod tests {
use super::*;
#[test]
fn test_middleware_result_type_is_result() {
// Verify type alias works
let _result: MiddlewareResult<()> = Ok(());
let _result: MiddlewareResult<()> = Err(HttpResponse::Unauthorized().finish());
}
#[test]
fn test_validate_and_rate_limit_signature() {
// Just verify the function signature is correct (compile-time test)
// Runtime tests require full AppState with mocks
let _ = validate_and_rate_limit;
}
}