diff --git a/Cargo.toml b/Cargo.toml index 3381c05..44a7c1a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -76,7 +76,8 @@ name = "proxy_bench" harness = false [features] -default = ["cli", "tls", "api"] +default = ["cli", "tls", "api", "logging"] cli = ["dep:clap", "dep:tracing-subscriber"] tls = ["dep:hyper-rustls"] api = ["dep:serde", "dep:serde_json"] +logging = [] diff --git a/README.md b/README.md index 827113a..c86b950 100644 --- a/README.md +++ b/README.md @@ -527,7 +527,7 @@ cargo run --example background - ⏳ TLS/SSL support - ⏳ WebSocket support - ⏳ Rate limiting -- ⏳ Request/response logging +- ✅ Structured access log with X-Request-ID (method, path, host, status, duration, bytes_sent) - ⏳ Metrics and monitoring ## Contributing diff --git a/src/api/endpoints.rs b/src/api/endpoints.rs index a8c5eff..db15ce8 100644 --- a/src/api/endpoints.rs +++ b/src/api/endpoints.rs @@ -29,7 +29,7 @@ where .status(200) .header("Content-Type", "application/json") .body(Full::new(Bytes::from(json))) - .unwrap(); + .expect("static response build"); Ok(response) } @@ -74,9 +74,9 @@ where .status(400) .header("Content-Type", "application/json") .body(Full::new(Bytes::from( - serde_json::to_string(&error_json).unwrap(), + serde_json::to_string(&error_json).expect("json!() is always valid"), ))) - .unwrap(); + .expect("static response build"); return Ok(response); } }; @@ -92,9 +92,9 @@ where .status(400) .header("Content-Type", "application/json") .body(Full::new(Bytes::from( - serde_json::to_string(&error_json).unwrap(), + serde_json::to_string(&error_json).expect("json!() is always valid"), ))) - .unwrap(); + .expect("static response build"); return Ok(response); } }; @@ -112,9 +112,9 @@ where .status(400) .header("Content-Type", "application/json") .body(Full::new(Bytes::from( - serde_json::to_string(&error_json).unwrap(), + serde_json::to_string(&error_json).expect("json!() is always valid"), ))) - .unwrap(); + .expect("static response build"); return Ok(response); } }; @@ -136,7 +136,7 @@ where .body(Full::new(Bytes::from( r#"{"status": "success", "message": "Configuration updated"}"#.to_string(), ))) - .unwrap(); + .expect("static response build"); Ok(response) } @@ -158,9 +158,9 @@ where .status(200) .header("Content-Type", "application/json") .body(Full::new(Bytes::from( - serde_json::to_string(&health).unwrap(), + serde_json::to_string(&health).expect("json!() is always valid"), ))) - .unwrap(); + .expect("static response build"); Ok(response) } diff --git a/src/api/middleware.rs b/src/api/middleware.rs index 7dd6abd..2d09028 100644 --- a/src/api/middleware.rs +++ b/src/api/middleware.rs @@ -32,7 +32,7 @@ fn unauthorized_response(message: &str) -> Response> { .status(StatusCode::UNAUTHORIZED) .header("Content-Type", "application/json") .body(Full::new(Bytes::from(body))) - .unwrap() + .expect("static response build") } /// Logging middleware diff --git a/src/api/server.rs b/src/api/server.rs index ad35655..715fef1 100644 --- a/src/api/server.rs +++ b/src/api/server.rs @@ -105,8 +105,8 @@ async fn handle_api_request( // 404 Not Found let response = Response::builder() .status(404) - .body(Full::new(bytes::Bytes::from("Not Found".to_string()))) - .unwrap(); + .body(Full::new(bytes::Bytes::from("Not Found"))) + .expect("static response build"); Ok(response) } } diff --git a/src/config/parser.rs b/src/config/parser.rs index 3a7e961..4c99db4 100644 --- a/src/config/parser.rs +++ b/src/config/parser.rs @@ -112,8 +112,10 @@ impl FromStr for Config { // 2. Handle closing brace if line == "}" { if directive_stack.len() > 1 { - let finished_directives = directive_stack.pop().unwrap(); - let block_info = block_stack.pop().unwrap(); + let finished_directives = directive_stack + .pop() + .expect("directive_stack has at least 2 elements"); + let block_info = block_stack.pop().expect("block_stack has matching entry"); let completed_directive = match block_info.directive_type.as_str() { "handle_path" => { @@ -145,12 +147,14 @@ impl FromStr for Config { directive_stack .last_mut() - .unwrap() + .expect("directive_stack has parent after pop") .push(completed_directive); } else { // Site block closed if let Some(address) = current_site_address.take() { - let site_directives = directive_stack.pop().unwrap(); + let site_directives = directive_stack + .pop() + .expect("site directive_stack is non-empty"); sites.insert( address.clone(), SiteConfig { @@ -309,7 +313,10 @@ impl FromStr for Config { } }; - directive_stack.last_mut().unwrap().push(directive); + directive_stack + .last_mut() + .expect("directive_stack is non-empty") + .push(directive); } Ok(Config { sites }) diff --git a/src/main.rs b/src/main.rs index 3f971a3..a80cd89 100644 --- a/src/main.rs +++ b/src/main.rs @@ -110,7 +110,7 @@ async fn run_with_api(cli: Cli, config: Config) -> Result<(), anyhow::Error> { // Wait for shutdown signal tokio::select! { // API server completed - api_result = async { api_handle.as_mut().unwrap().await } => { + api_result = async { api_handle.as_mut().expect("api_handle is Some").await } => { match api_result { Ok(Ok(())) => info!("API server shut down gracefully"), Ok(Err(e)) => error!("API server error: {}", e), @@ -120,7 +120,7 @@ async fn run_with_api(cli: Cli, config: Config) -> Result<(), anyhow::Error> { let _ = shutdown_tx.send(()); }, // Proxy server completed - proxy_result = async { proxy_handle.as_mut().unwrap().await } => { + proxy_result = async { proxy_handle.as_mut().expect("proxy_handle is Some").await } => { match proxy_result { Ok(Ok(())) => info!("Proxy server shut down gracefully"), Ok(Err(e)) => error!("Proxy server error: {}", e), @@ -142,7 +142,12 @@ async fn run_with_api(cli: Cli, config: Config) -> Result<(), anyhow::Error> { let timeout = tokio::time::Duration::from_secs(30); // Wait for API server - match tokio::time::timeout(timeout, api_handle.take().unwrap()).await { + match tokio::time::timeout( + timeout, + api_handle.take().expect("api_handle consumed once"), + ) + .await + { Ok(Ok(Ok(()))) => info!("API server shut down"), Ok(Ok(Err(e))) => warn!("API server shutdown error: {}", e), Ok(Err(e)) => warn!("API server task error: {}", e), @@ -155,7 +160,12 @@ async fn run_with_api(cli: Cli, config: Config) -> Result<(), anyhow::Error> { } // Wait for proxy server - match tokio::time::timeout(timeout, proxy_handle.take().unwrap()).await { + match tokio::time::timeout( + timeout, + proxy_handle.take().expect("proxy_handle consumed once"), + ) + .await + { Ok(Ok(Ok(()))) => info!("Proxy server shut down"), Ok(Ok(Err(e))) => warn!("Proxy server shutdown error: {}", e), Ok(Err(e)) => warn!("Proxy server task error: {}", e), diff --git a/src/proxy/access_log.rs b/src/proxy/access_log.rs new file mode 100644 index 0000000..2e360df --- /dev/null +++ b/src/proxy/access_log.rs @@ -0,0 +1,346 @@ +use hyper::Request; + +#[cfg(feature = "logging")] +use std::time::Instant; +#[cfg(feature = "logging")] +use tracing::info; + +/// Generate or extract a request ID for tracing. +/// +/// If the incoming request already has an `X-Request-ID` header, reuse it. +/// Otherwise, generate a new UUID v4 and inject it into the request. +pub fn ensure_request_id(req: &mut Request) -> String { + use hyper::header::HeaderValue; + + if let Some(existing) = req.headers().get("X-Request-ID").cloned() { + if let Ok(id) = existing.to_str() { + return id.to_string(); + } + } + + // Generate new UUID v4 + let id = uuid::Uuid::new_v4().to_string(); + if let Ok(val) = HeaderValue::from_str(&id) { + req.headers_mut().insert("X-Request-ID", val); + } + id +} + +/// Read the final request ID from headers after directive processing. +/// +/// This resolves the conflict where a `header X-Request-ID {uuid}` directive +/// may have overwritten the initially generated ID. +pub fn final_request_id(req: &Request, fallback: &str) -> String { + req.headers() + .get("X-Request-ID") + .and_then(|v| v.to_str().ok()) + .unwrap_or(fallback) + .to_string() +} + +/// Write a structured access log entry via tracing. +/// +/// Produces a structured log line with all request metadata: +/// ```text +/// access_log req_id=a1b2c3 remote=127.0.0.1 method=GET path=/api/users host=localhost:8080 status=200 duration_ms=1.23 bytes_sent=1234 +/// ``` +/// # Semantics +/// +/// - **`duration_ms`**: Measures time from request entry to response headers ready +/// (i.e., TTFB — Time To First Byte). For reverse proxy, this includes the +/// backend round-trip but **not** the streaming of the response body to the client. +/// For error responses and direct responses, it effectively covers the full handling time. +/// End-to-end connection duration (including full body transfer) is not currently tracked. +/// +/// - **`bytes_sent`**: `None` (`"-"` in log) for streaming responses (reverse proxy), +/// `Some(n)` for direct responses where the body size is known exactly. +#[cfg(feature = "logging")] +#[allow(clippy::too_many_arguments)] +pub fn log_access( + request_id: &str, + remote_addr: std::net::SocketAddr, + method: &str, + path: &str, + host: &str, + status: u16, + duration_ms: f64, + bytes_sent: Option, +) { + info!( + req_id = %request_id, + remote = %remote_addr.ip(), + method = %method, + path = %path, + host = %host, + status = status, + duration_ms = format_args!("{:.2}", duration_ms), + bytes_sent = bytes_sent.map(|n| n.to_string()).unwrap_or_else(|| "-".to_string()), + "access_log" + ); +} + +/// RAII guard for access logging. +/// +/// Captures request metadata on creation and writes the access log on drop. +/// If `finish()` was called, uses the provided status. Otherwise logs 500 +/// (covers panics and unexpected error paths). +/// +/// Only available when the `logging` feature is enabled. +/// +/// # Usage +/// ```ignore +/// let mut guard = AccessLogGuard::new(request_id, remote_addr, method, path, host); +/// // ... handle request ... +/// guard.finish(200); +/// guard.set_bytes_sent(body.len()); +/// // log is written when guard is dropped +/// ``` +#[cfg(feature = "logging")] +pub struct AccessLogGuard { + request_id: String, + remote_addr: std::net::SocketAddr, + method: String, + path: String, + host: String, + start: Instant, + status: Option, + bytes_sent: Option, +} + +#[cfg(feature = "logging")] +impl AccessLogGuard { + pub fn new( + request_id: String, + remote_addr: std::net::SocketAddr, + method: String, + path: String, + host: String, + ) -> Self { + Self { + request_id, + remote_addr, + method, + path, + host, + start: Instant::now(), + status: None, + bytes_sent: None, + } + } + + /// Set the final HTTP status code. Will be used in the access log. + pub fn finish(&mut self, status: u16) { + self.status = Some(status); + } + + /// Set the number of bytes sent in the response body. + /// Use for direct responses (Respond, Redirect, errors) where body size is known. + /// For streaming proxy responses, leave as None (logged as `-`). + pub fn set_bytes_sent(&mut self, bytes: usize) { + self.bytes_sent = Some(bytes); + } + + /// Get the current request ID. + /// + /// Note: handler.rs tracks request_id as a local variable for use in responses, + /// so this accessor is not called internally. Kept for external API completeness. + #[allow(dead_code)] + pub fn request_id(&self) -> &str { + &self.request_id + } + + /// Update the request ID (e.g., after directive processing changed X-Request-ID). + pub fn set_request_id(&mut self, id: String) { + self.request_id = id; + } +} + +#[cfg(feature = "logging")] +impl Drop for AccessLogGuard { + fn drop(&mut self) { + let status = self.status.unwrap_or(500); + let duration_ms = self.start.elapsed().as_secs_f64() * 1000.0; + log_access( + &self.request_id, + self.remote_addr, + &self.method, + &self.path, + &self.host, + status, + duration_ms, + self.bytes_sent, + ); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use bytes::Bytes; + use http_body_util::Empty; + + fn make_request() -> Request> { + Request::builder() + .method("GET") + .uri("/test") + .body(Empty::new()) + .unwrap() + } + + #[test] + fn test_ensure_request_id_generates_when_missing() { + let mut req = make_request(); + assert!(req.headers().get("X-Request-ID").is_none()); + + let id = ensure_request_id(&mut req); + + assert!(!id.is_empty()); + assert_eq!(id.len(), 36, "Should be a UUID"); + assert!(req.headers().get("X-Request-ID").is_some()); + } + + #[test] + fn test_ensure_request_id_reuses_existing() { + let mut req = Request::builder() + .header("X-Request-ID", "my-custom-id") + .body(Empty::::new()) + .unwrap(); + + let id = ensure_request_id(&mut req); + + assert_eq!(id, "my-custom-id"); + } + + #[test] + fn test_final_request_id_after_directive_change() { + let mut req = make_request(); + let original_id = ensure_request_id(&mut req); + + // Simulate a directive changing the header + req.headers_mut().insert( + "X-Request-ID", + hyper::header::HeaderValue::from_static("directive-id"), + ); + + let final_id = final_request_id(&req, &original_id); + assert_eq!(final_id, "directive-id"); + } + + #[test] + fn test_final_request_id_fallback_when_no_header() { + let req = make_request(); + let final_id = final_request_id(&req, "fallback-id"); + assert_eq!(final_id, "fallback-id"); + } + + #[cfg(feature = "logging")] + mod logging_tests { + use super::*; + + #[test] + fn test_log_access_does_not_panic() { + let addr: std::net::SocketAddr = "127.0.0.1:54321".parse().unwrap(); + log_access( + "abc123", + addr, + "GET", + "/api/users", + "localhost:8080", + 200, + 1.23, + Some(1234), + ); + } + + #[test] + fn test_log_access_streaming_response() { + let addr: std::net::SocketAddr = "127.0.0.1:54321".parse().unwrap(); + // bytes_sent = None for streaming (reverse proxy) + log_access( + "abc123", + addr, + "GET", + "/stream", + "localhost:8080", + 200, + 50.5, + None, + ); + } + + #[test] + fn test_log_access_error_status() { + let addr: std::net::SocketAddr = "10.0.0.1:12345".parse().unwrap(); + log_access( + "def456", + addr, + "POST", + "/api/orders", + "api.example.com", + 502, + 30001.5, + Some(0), + ); + } + + #[test] + fn test_access_log_guard_finish() { + let addr: std::net::SocketAddr = "127.0.0.1:12345".parse().unwrap(); + let mut guard = AccessLogGuard::new( + "test-id".to_string(), + addr, + "GET".to_string(), + "/test".to_string(), + "localhost".to_string(), + ); + guard.finish(200); + guard.set_bytes_sent(1024); + // Drop will log — just verify no panic + } + + #[test] + fn test_access_log_guard_default_500() { + let addr: std::net::SocketAddr = "127.0.0.1:12345".parse().unwrap(); + let guard = AccessLogGuard::new( + "test-id".to_string(), + addr, + "GET".to_string(), + "/test".to_string(), + "localhost".to_string(), + ); + // No finish() called — drop logs 500 with bytes_sent=None + drop(guard); + } + + #[test] + fn test_access_log_guard_set_request_id() { + let addr: std::net::SocketAddr = "127.0.0.1:12345".parse().unwrap(); + let mut guard = AccessLogGuard::new( + "old-id".to_string(), + addr, + "GET".to_string(), + "/test".to_string(), + "localhost".to_string(), + ); + assert_eq!(guard.request_id(), "old-id"); + guard.set_request_id("new-id".to_string()); + assert_eq!(guard.request_id(), "new-id"); + guard.finish(200); + } + + #[test] + fn test_access_log_guard_bytes_sent() { + let addr: std::net::SocketAddr = "127.0.0.1:12345".parse().unwrap(); + let mut guard = AccessLogGuard::new( + "test-id".to_string(), + addr, + "POST".to_string(), + "/submit".to_string(), + "localhost".to_string(), + ); + guard.finish(201); + guard.set_bytes_sent(42); + // Drop logs with bytes_sent=Some(42) + } + } +} diff --git a/src/proxy/handler.rs b/src/proxy/handler.rs index b9fbaae..e858ebc 100644 --- a/src/proxy/handler.rs +++ b/src/proxy/handler.rs @@ -11,7 +11,15 @@ use std::sync::Arc; use tokio::time::{timeout, Duration}; use tracing::{error, info}; +#[cfg(feature = "logging")] +use tracing::info_span; +#[cfg(feature = "logging")] +use tracing::Instrument; + use crate::config::Config; +#[cfg(feature = "logging")] +use crate::proxy::access_log::AccessLogGuard; +use crate::proxy::access_log::{ensure_request_id, final_request_id}; use crate::proxy::ActionResult; use crate::proxy::directives::{ @@ -41,8 +49,12 @@ fn is_hop_header(name: &header::HeaderName) -> bool { ) } -/// Process directives in order, applying modifications and returning final action -/// Supports recursive handling of handle_path blocks +/// Process directives in order, applying modifications and returning final action. +/// Supports recursive handling of handle_path blocks. +/// +/// Note: `info!` logs here are correlated with request ID only when the `logging` +/// feature is enabled (via the tracing span set in `proxy()`). Without `logging`, +/// these logs appear without request context. pub fn process_directives( directives: &[crate::config::Directive], req: &mut Request, @@ -52,60 +64,48 @@ pub fn process_directives( for directive in directives { match directive { - // Apply header modifications using directive handler - - // Apply header modifications using directive handler crate::config::Directive::Header { name, value } => { if let Err(e) = handle_header(name, value.as_deref(), req) { info!(" Failed to apply header {}: {}", name, e); } } - // Apply URI replacements using directive handler crate::config::Directive::UriReplace { find, replace } => { handle_uri_replace(find, replace, &mut modified_path); } - // Strip prefix from URI path crate::config::Directive::StripPrefix { prefix } => { handle_strip_prefix(prefix, &mut modified_path); } - // Handle path-based routing recursively crate::config::Directive::HandlePath { pattern, directives: nested_directives, } => { if let Some(remaining_path) = match_pattern(pattern, &modified_path) { info!(" Matched handle_path: {}", pattern); - // Recursively process nested directives with remaining path return process_directives(nested_directives, req, &remaining_path); } } - // Method-based directives crate::config::Directive::Method { methods, directives: nested_directives, } => { if handle_method(methods, req) { info!(" Matched method directive"); - // Process nested directives with same path return process_directives(nested_directives, req, &modified_path); } } - // Redirect - return redirect response with Location header crate::config::Directive::Redirect { status, url } => { return Ok(handle_redirect(status, url)); } - // Direct response - return immediately using directive handler crate::config::Directive::Respond { status, body } => { return Ok(handle_respond(status, body)); } - // Reverse proxy - return action using directive handler crate::config::Directive::ReverseProxy { to, connect_timeout, @@ -143,204 +143,273 @@ pub async fn proxy( config: Arc, remote_addr: std::net::SocketAddr, ) -> Result, Error> { - // Get path from URI (using String to avoid borrow conflict with mutable req) - let path = req.uri().path().to_string(); + // Generate or reuse request ID + let initial_request_id = ensure_request_id(&mut req); - // Get host from Host header (includes port, e.g., "localhost:8080") + // Extract request info before processing + #[cfg(feature = "logging")] + let method = req.method().clone().to_string(); + let path = req.uri().path().to_string(); let host = req .headers() .get(hyper::header::HOST) .and_then(|h| h.to_str().ok()) - .unwrap_or("localhost"); - - // Logging with enabled check to avoid string formatting when disabled - if tracing::enabled!(tracing::Level::INFO) { - // Removed info logging from hot path for performance - // Use DEBUG level if needed for troubleshooting - } - - // Find site configuration by host (with port!) - let site_config = match config.sites.get(host) { - Some(config) => config, - None => { - error!("No configuration found for host: {}", host); - return Ok(error_response( - StatusCode::NOT_FOUND, - &format!("No configuration found for host: {}", host), - )); - } - }; - - // Process directives in correct order - let action_result = - process_directives(&site_config.directives, &mut req, &path).map_err(anyhow::Error::msg)?; - - // Execute action - match action_result { - ActionResult::Redirect { status, url } => { - let status_code = StatusCode::from_u16(status).unwrap_or(StatusCode::FOUND); - let boxed: ResponseBody = Full::new(Bytes::from(url.clone())) - .map_err(|e| Box::new(e) as Box) - .boxed(); - Ok(Response::builder() - .status(status_code) - .header("Location", &url) - .body(boxed)?) - } - ActionResult::Respond { status, body } => { - let status_code = StatusCode::from_u16(status).unwrap_or(StatusCode::OK); - let boxed: ResponseBody = Full::new(Bytes::from(body)) - .map_err(|e| Box::new(e) as Box) - .boxed(); - Ok(Response::builder().status(status_code).body(boxed)?) - } - ActionResult::ReverseProxy { - backend_url, - path_to_send, - connect_timeout: _, - read_timeout, - } => { - // Add protocol if missing - let backend_with_proto = - if backend_url.starts_with("http://") || backend_url.starts_with("https://") { - backend_url - } else { - format!("http://{}", backend_url) - }; - - // Use Uri::from_parts() instead of format!() + parse() - faster! - let mut parts = backend_with_proto.parse::()?.into_parts(); - parts.path_and_query = Some(path_to_send.parse()?); - let new_uri = Uri::from_parts(parts)?; - - // Logging with enabled check to avoid string formatting when disabled - if tracing::enabled!(tracing::Level::INFO) { - // Removed info logging from hot path for performance - // Use DEBUG level if needed for troubleshooting - } - - *req.uri_mut() = new_uri.clone(); - - // Save original host for X-Forwarded headers - // Clone HeaderValue directly - 0 allocations! - let original_host_header = req.headers().get(hyper::header::HOST).cloned(); - - // Update Host header for backend - req.headers_mut().remove(hyper::header::HOST); - if let Some(authority) = new_uri.authority() { - if let Ok(host_value) = authority.as_str().parse::() { - req.headers_mut().insert(hyper::header::HOST, host_value); + .unwrap_or("localhost") + .to_string(); + + #[cfg(feature = "logging")] + let span = info_span!("request", req_id = %initial_request_id); + + #[allow(unused_variables)] + let future = async move { + #[cfg(feature = "logging")] + let mut log_guard = AccessLogGuard::new( + initial_request_id.clone(), + remote_addr, + method, + path.clone(), + host.clone(), + ); + + // Find site configuration by host (with port!) + let site_config = match config.sites.get(&host) { + Some(config) => config, + None => { + error!("No configuration found for host: {}", host); + let (response, _body_len) = error_response_with_id( + StatusCode::NOT_FOUND, + &format!("No configuration found for host: {}", host), + &initial_request_id, + ); + #[cfg(feature = "logging")] + { + log_guard.set_bytes_sent(_body_len); + log_guard.finish(404); } + return Ok(response); } - - // Add X-Forwarded-* headers for backend visibility - // X-Forwarded-Host: original Host header from client - if let Some(host_value) = original_host_header.clone() { - req.headers_mut().insert("X-Forwarded-Host", host_value); + }; + + // Process directives in correct order + let action_result = match process_directives(&site_config.directives, &mut req, &path) { + Ok(result) => result, + Err(e) => { + error!("Directive processing error: {}", e); + let final_id = final_request_id(&req, &initial_request_id); + #[cfg(feature = "logging")] + { + log_guard.set_request_id(final_id.clone()); + tracing::Span::current().record("req_id", final_id.as_str()); + } + let (response, _body_len) = + error_response_with_id(StatusCode::INTERNAL_SERVER_ERROR, &e, &final_id); + #[cfg(feature = "logging")] + { + log_guard.set_bytes_sent(_body_len); + log_guard.finish(500); + } + return Ok(response); } + }; + + // Read final request ID (directive may have overwritten X-Request-ID) + let request_id = final_request_id(&req, &initial_request_id); + #[cfg(feature = "logging")] + { + log_guard.set_request_id(request_id.clone()); + // Update the tracing span with the final req_id + tracing::Span::current().record("req_id", request_id.as_str()); + } - // X-Forwarded-Proto: scheme from original request (http or https) - let original_scheme = req.uri().scheme_str().unwrap_or("http"); - // Use from_static for known values - 0 allocations! - match original_scheme { - "http" => { - req.headers_mut().insert( - "X-Forwarded-Proto", - hyper::header::HeaderValue::from_static("http"), - ); + // Execute action + match action_result { + ActionResult::Redirect { status, url } => { + let status_code = StatusCode::from_u16(status).unwrap_or(StatusCode::FOUND); + + let boxed: ResponseBody = Full::new(Bytes::new()) + .map_err(|e| Box::new(e) as Box) + .boxed(); + let response = Response::builder() + .status(status_code) + .header("Location", &url) + .header("X-Request-ID", &request_id) + .body(boxed)?; + #[cfg(feature = "logging")] + { + log_guard.set_bytes_sent(0); + log_guard.finish(status_code.as_u16()); } - "https" => { - req.headers_mut().insert( - "X-Forwarded-Proto", - hyper::header::HeaderValue::from_static("https"), - ); + Ok(response) + } + ActionResult::Respond { status, body } => { + let status_code = StatusCode::from_u16(status).unwrap_or(StatusCode::OK); + let _body_len = body.len(); + + let boxed: ResponseBody = Full::new(Bytes::from(body)) + .map_err(|e| Box::new(e) as Box) + .boxed(); + let response = Response::builder() + .status(status_code) + .header("X-Request-ID", &request_id) + .body(boxed)?; + #[cfg(feature = "logging")] + { + log_guard.set_bytes_sent(_body_len); + log_guard.finish(status_code.as_u16()); } - _ => {} // ignore unknown schemes + Ok(response) } + ActionResult::ReverseProxy { + backend_url, + path_to_send, + connect_timeout: _, + read_timeout, + } => { + // Add protocol if missing + let backend_with_proto = + if backend_url.starts_with("http://") || backend_url.starts_with("https://") { + backend_url + } else { + format!("http://{}", backend_url) + }; - // X-Forwarded-For: real client IP from TCP connection - if let Ok(ip_value) = - hyper::header::HeaderValue::from_str(&remote_addr.ip().to_string()) - { - req.headers_mut().insert("X-Forwarded-For", ip_value); - } + // Use Uri::from_parts() instead of format!() + parse() - faster! + let mut parts = backend_with_proto.parse::()?.into_parts(); + parts.path_and_query = Some(path_to_send.parse()?); + let new_uri = Uri::from_parts(parts)?; - // Remove hop-by-hop headers from request before sending to backend - // Connection header must not be proxied (hyper manages connections) - req.headers_mut().remove(header::CONNECTION); - - // Remove Accept-Encoding to prevent compression - // Compression breaks streaming and SSE - req.headers_mut().remove("accept-encoding"); - - // Forward request to backend with configurable timeout (default 30s) - let backend_timeout = read_timeout.unwrap_or(30); - match timeout(Duration::from_secs(backend_timeout), client.request(req)).await { - Ok(Ok(response)) => { - // Successfully received response from backend - let status = response.status(); - let headers = response.headers().clone(); - - // Logging with enabled check to avoid string formatting when disabled - if tracing::enabled!(tracing::Level::INFO) { - // Removed info logging from hot path for performance - // Use DEBUG level if needed for troubleshooting - } + *req.uri_mut() = new_uri.clone(); - // Stream response body directly (no buffering) - let mut builder = Response::builder().status(status); + // Save original host for X-Forwarded headers + let original_host_header = req.headers().get(hyper::header::HOST).cloned(); - // Copy all headers from backend, filtering out hop-by-hop headers - // Hop-by-hop headers should not be proxied per RFC 7230 - // Also remove Content-Length to let hyper handle chunked encoding - for (name, value) in headers.iter() { - if !is_hop_header(name) && name != header::CONTENT_LENGTH { - builder = builder.header(name, value); - } + // Update Host header for backend + req.headers_mut().remove(hyper::header::HOST); + if let Some(authority) = new_uri.authority() { + if let Ok(host_value) = authority.as_str().parse::() + { + req.headers_mut().insert(hyper::header::HOST, host_value); } + } - // Extract streaming body and convert to BoxBody - let (_, incoming_body) = response.into_parts(); - let boxed: ResponseBody = incoming_body - .map_err(|e| Box::new(e) as Box) - .boxed(); - - Ok(builder.body(boxed)?) + // Add X-Forwarded-* headers for backend visibility + if let Some(host_value) = original_host_header.clone() { + req.headers_mut().insert("X-Forwarded-Host", host_value); } - Ok(Err(e)) => { - // Backend unavailable - return 502 Bad Gateway - error!("Backend connection failed: {:?}", e); - if e.is_connect() { - error!(" Reason: Connection refused - backend unavailable"); - } else { - error!(" Reason: Other connection error"); + // X-Forwarded-Proto: scheme from original request + let original_scheme = req.uri().scheme_str().unwrap_or("http"); + match original_scheme { + "http" => { + req.headers_mut().insert( + "X-Forwarded-Proto", + hyper::header::HeaderValue::from_static("http"), + ); + } + "https" => { + req.headers_mut().insert( + "X-Forwarded-Proto", + hyper::header::HeaderValue::from_static("https"), + ); } + _ => {} + } - Ok(error_response( - StatusCode::BAD_GATEWAY, - "Backend service unavailable", - )) + // X-Forwarded-For: real client IP + if let Ok(ip_value) = + hyper::header::HeaderValue::from_str(&remote_addr.ip().to_string()) + { + req.headers_mut().insert("X-Forwarded-For", ip_value); } - Err(_) => { - // Timeout - return 504 Gateway Timeout - error!( - "Backend request timed out after {} seconds", - backend_timeout - ); - - Ok(error_response( - StatusCode::GATEWAY_TIMEOUT, - "Backend request timed out", - )) + + // Remove hop-by-hop headers + req.headers_mut().remove(header::CONNECTION); + req.headers_mut().remove("accept-encoding"); + + // Forward request to backend with configurable timeout (default 30s) + let backend_timeout = read_timeout.unwrap_or(30); + match timeout(Duration::from_secs(backend_timeout), client.request(req)).await { + Ok(Ok(response)) => { + let status = response.status(); + let headers = response.headers().clone(); + + // Stream response body directly (no buffering) + let mut builder = Response::builder().status(status); + + // Copy headers, filtering hop-by-hop + for (name, value) in headers.iter() { + if !is_hop_header(name) && name != header::CONTENT_LENGTH { + builder = builder.header(name, value); + } + } + + let (_, incoming_body) = response.into_parts(); + let boxed: ResponseBody = incoming_body + .map_err(|e| Box::new(e) as Box) + .boxed(); + + let response = builder.header("X-Request-ID", &request_id).body(boxed)?; + #[cfg(feature = "logging")] + log_guard.finish(status.as_u16()); + Ok(response) + } + Ok(Err(e)) => { + error!("Backend connection failed: {:?}", e); + if e.is_connect() { + error!(" Reason: Connection refused - backend unavailable"); + } else { + error!(" Reason: Other connection error"); + } + + let (response, _body_len) = error_response_with_id( + StatusCode::BAD_GATEWAY, + "Backend service unavailable", + &request_id, + ); + #[cfg(feature = "logging")] + { + log_guard.set_bytes_sent(_body_len); + log_guard.finish(502); + } + Ok(response) + } + Err(_) => { + error!( + "Backend request timed out after {} seconds", + backend_timeout + ); + + let (response, _body_len) = error_response_with_id( + StatusCode::GATEWAY_TIMEOUT, + "Backend request timed out", + &request_id, + ); + #[cfg(feature = "logging")] + { + log_guard.set_bytes_sent(_body_len); + log_guard.finish(504); + } + Ok(response) + } } } } - } + }; + + #[cfg(feature = "logging")] + let future = future.instrument(span); + + future.await } -/// Creates HTTP response with error -fn error_response(status: StatusCode, message: &str) -> Response { +/// Creates HTTP response with error and X-Request-ID header +/// +/// Returns both the response and the body length (for access logging). +fn error_response_with_id( + status: StatusCode, + message: &str, + request_id: &str, +) -> (Response, usize) { let body = format!( r#" @@ -359,16 +428,29 @@ fn error_response(status: StatusCode, message: &str) -> Response { message ); + let body_len = body.len(); let full = Full::new(Bytes::from(body)); let boxed: ResponseBody = full .map_err(|e| Box::new(e) as Box) .boxed(); - Response::builder() + let mut builder = Response::builder() .status(status) - .header("Content-Type", "text/html; charset=utf-8") - .body(boxed) - .unwrap() + .header("Content-Type", "text/html; charset=utf-8"); + + if let Ok(val) = hyper::header::HeaderValue::from_str(request_id) { + builder = builder.header("X-Request-ID", val); + } + + let response = builder.body(boxed).unwrap_or_else(|_| { + Response::new( + Full::new(Bytes::from("Internal Server Error")) + .map_err(|e| Box::new(e) as Box) + .boxed(), + ) + }); + + (response, body_len) } /// Match path against pattern (supports wildcard *) @@ -376,14 +458,13 @@ fn error_response(status: StatusCode, message: &str) -> Response { pub fn match_pattern(pattern: &str, path: &str) -> Option { if let Some(prefix) = pattern.strip_suffix("/*") { if path.starts_with(prefix) { - // Remove prefix and return remaining path let remaining = path.strip_prefix(prefix).unwrap_or(path); Some(remaining.to_string()) } else { None } } else if pattern == path { - Some("/".to_string()) // Exact match, send root + Some("/".to_string()) } else { None } diff --git a/src/proxy/mod.rs b/src/proxy/mod.rs index c0433c8..65af6df 100644 --- a/src/proxy/mod.rs +++ b/src/proxy/mod.rs @@ -1,3 +1,4 @@ +mod access_log; mod directives; pub mod handler; #[allow(clippy::module_inception)] diff --git a/src/proxy/proxy.rs b/src/proxy/proxy.rs index 0709701..a95adab 100644 --- a/src/proxy/proxy.rs +++ b/src/proxy/proxy.rs @@ -94,7 +94,7 @@ impl Proxy { http.set_nodelay(true); let https = HttpsConnectorBuilder::new() .with_native_roots() - .unwrap() + .expect("Failed to load native TLS root certificates") .https_or_http() .enable_http1() .wrap_connector(http); @@ -139,7 +139,7 @@ impl Proxy { http.set_nodelay(true); let https = HttpsConnectorBuilder::new() .with_native_roots() - .unwrap() + .expect("Failed to load native TLS root certificates") .https_or_http() .enable_http1() .wrap_connector(http);