diff --git a/Cargo.lock b/Cargo.lock index ad91b208..06515bcf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7952,6 +7952,7 @@ dependencies = [ "base64 0.23.1", "bytes", "chrono", + "chrono-tz", "criterion", "dashmap 7.0.0-rc2", "failsafe", diff --git a/Makefile b/Makefile index 07e66485..13d56adb 100644 --- a/Makefile +++ b/Makefile @@ -78,7 +78,7 @@ set +a; \ export SYNCTV_DATA_DIR="$(DEV_DATA_DIR)"; \ export SYNCTV_DATABASE_URL="$(DEV_DATABASE_URL)"; \ export SYNCTV_REDIS_URL="$(DEV_REDIS_URL)"; \ -export SYNCTV_LOGGING_LEVEL="$${SYNCTV_LOGGING_LEVEL:-debug}"; \ +export SYNCTV_LOGGING_LEVEL="$${SYNCTV_LOGGING_LEVEL:-info}"; \ export SYNCTV_SERVER_LOGGING_LEVEL="$${SYNCTV_SERVER_LOGGING_LEVEL:-debug}"; \ export SYNCTV_LIVESTREAM_LOGGING_LEVEL="$${SYNCTV_LIVESTREAM_LOGGING_LEVEL:-debug}"; \ export SYNCTV_WEBRTC_LOGGING_LEVEL="$${SYNCTV_WEBRTC_LOGGING_LEVEL:-debug}"; \ diff --git a/synctv-api-common/src/api_runtime.rs b/synctv-api-common/src/api_runtime.rs index b4edfa33..d646c40d 100644 --- a/synctv-api-common/src/api_runtime.rs +++ b/synctv-api-common/src/api_runtime.rs @@ -226,6 +226,7 @@ impl Default for ConnectionLimitSettings { #[derive(Clone, Debug)] pub struct ApiRuntimeSettings { pub server: ApiServerSettings, + pub access_log: crate::AccessLogSettings, pub request_rate_limits: RequestRateLimitSettings, pub metrics: MetricsRuntimeSettings, pub cluster_enabled: bool, @@ -243,6 +244,7 @@ impl Default for ApiRuntimeSettings { fn default() -> Self { Self { server: ApiServerSettings::default(), + access_log: crate::AccessLogSettings::default(), request_rate_limits: RequestRateLimitSettings::default(), metrics: MetricsRuntimeSettings::default(), cluster_enabled: false, diff --git a/synctv-api-common/src/lib.rs b/synctv-api-common/src/lib.rs index 8ea30820..7abc2a49 100644 --- a/synctv-api-common/src/lib.rs +++ b/synctv-api-common/src/lib.rs @@ -39,6 +39,7 @@ pub mod status; pub mod synology_image_urls; #[cfg(any(test, feature = "test-support"))] pub mod test_support; +pub mod transport_access_log; pub mod webrtc_status; pub use api_runtime::*; @@ -51,6 +52,7 @@ pub use realtime_fanout::*; pub use runtime::RealtimeAdmissionError; pub use runtime_adapters::proxy_slice_cache_options_from_runtime_settings; pub use server_settings::{ - validate_cors_origin, AndroidAppAssociationSettings, ApiServerSettings, DEFAULT_PROJECT_URL, + validate_cors_origin, AccessLogSettings, AndroidAppAssociationSettings, ApiServerSettings, + DEFAULT_PROJECT_URL, }; pub use synctv_adapter::{PublicIdCodec, PublicIdConfig, PublicIdKind, PublicIdType}; diff --git a/synctv-api-common/src/observability/metrics.rs b/synctv-api-common/src/observability/metrics.rs index 1833648a..f4cd53aa 100644 --- a/synctv-api-common/src/observability/metrics.rs +++ b/synctv-api-common/src/observability/metrics.rs @@ -6,106 +6,10 @@ pub use synctv_core::metrics::http::{ HTTP_REQUESTS_IN_FLIGHT, HTTP_REQUESTS_TOTAL, HTTP_REQUEST_DURATION_SECONDS, }; +pub use synctv_core::metrics::remote_transport::{ + REMOTE_TRANSPORT_REQUESTS_TOTAL, REMOTE_TRANSPORT_REQUEST_DURATION, +}; pub use synctv_core::metrics::livestream::LIVESTREAM_FLV_SLOW_CLIENT_TERMINATIONS_TOTAL; pub use synctv_core::metrics::gather_metrics; - -/// Normalize an HTTP request path for metric labels. -/// -/// Replaces route parameters and dynamic IDs with placeholders to avoid -/// high-cardinality labels. -#[must_use] -pub fn normalize_path(path: &str) -> String { - let segments: Vec<&str> = path.split('/').collect(); - let mut result = Vec::with_capacity(segments.len()); - - for (i, segment) in segments.iter().enumerate() { - if segment.is_empty() { - result.push(*segment); - continue; - } - - let prev = if i > 0 { segments.get(i - 1) } else { None }; - let is_id = matches!( - prev, - Some( - &"rooms" - | &"media" - | &"chat" - | &"playlists" - | &"users" - | &"notifications" - | &"settings" - | &"members" - ) - ); - - if is_id || is_dynamic_segment(segment) { - result.push(":id"); - } else { - result.push(segment); - } - } - - result.join("/") -} - -fn is_dynamic_segment(segment: &str) -> bool { - if segment.len() == 36 && segment.chars().all(|c| c.is_ascii_hexdigit() || c == '-') { - let parts: Vec<&str> = segment.split('-').collect(); - if parts.len() == 5 - && parts[0].len() == 8 - && parts[1].len() == 4 - && parts[2].len() == 4 - && parts[3].len() == 4 - && parts[4].len() == 12 - { - return true; - } - } - - if segment.chars().all(|c| c.is_ascii_digit()) && !segment.is_empty() { - return true; - } - - if segment.len() == 32 && segment.chars().all(|c| c.is_ascii_hexdigit()) { - return true; - } - - false -} - -#[cfg(test)] -mod tests { - use super::normalize_path; - - #[test] - fn normalize_path_existing_resources() { - assert_eq!( - normalize_path("/api/rooms/abc123/media"), - "/api/rooms/:id/media" - ); - assert_eq!(normalize_path("/api/media/xyz789"), "/api/media/:id"); - assert_eq!(normalize_path("/api/chat/msg001"), "/api/chat/:id"); - assert_eq!(normalize_path("/api/playlists/pl123"), "/api/playlists/:id"); - } - - #[test] - fn normalize_path_extended_resources() { - assert_eq!(normalize_path("/api/users/u123"), "/api/users/:id"); - assert_eq!( - normalize_path("/api/notifications/n456"), - "/api/notifications/:id" - ); - assert_eq!(normalize_path("/api/settings/s789"), "/api/settings/:id"); - assert_eq!(normalize_path("/api/members/m012"), "/api/members/:id"); - } - - #[test] - fn normalize_path_without_id_segments() { - assert_eq!(normalize_path("/api/rooms"), "/api/rooms"); - assert_eq!(normalize_path("/api/health"), "/api/health"); - assert_eq!(normalize_path("/metrics"), "/metrics"); - } -} diff --git a/synctv-api-common/src/server_settings.rs b/synctv-api-common/src/server_settings.rs index a73abb71..757d1770 100644 --- a/synctv-api-common/src/server_settings.rs +++ b/synctv-api-common/src/server_settings.rs @@ -6,6 +6,21 @@ pub struct AndroidAppAssociationSettings { pub sha256_cert_fingerprints: Vec, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct AccessLogSettings { + pub enabled: bool, + pub slow_request_threshold_ms: u64, +} + +impl Default for AccessLogSettings { + fn default() -> Self { + Self { + enabled: true, + slow_request_threshold_ms: 1_000, + } + } +} + #[derive(Debug, Clone)] pub struct ApiServerSettings { pub bind_address: String, diff --git a/synctv-api-common/src/transport_access_log.rs b/synctv-api-common/src/transport_access_log.rs new file mode 100644 index 00000000..18834490 --- /dev/null +++ b/synctv-api-common/src/transport_access_log.rs @@ -0,0 +1,1104 @@ +use std::{ + net::{IpAddr, SocketAddr}, + pin::Pin, + task::{Context, Poll}, + time::{Duration, Instant}, +}; + +use axum::{ + body::{Body, Bytes, HttpBody}, + extract::{ConnectInfo, MatchedPath, Request}, + http::{header::ToStrError, HeaderMap, HeaderName, HeaderValue, Method, StatusCode}, + middleware::Next, + response::Response, +}; +use hyper::body::{Frame, SizeHint}; + +use crate::{ + observability::metrics, request_context::CURRENT_REQUEST_ID, AccessLogSettings, + ApiServerSettings, +}; + +const ACCESS_LOG_TARGET: &str = "synctv::access"; +const MAX_LOGGED_ROUTE_LEN: usize = 512; +const UNKNOWN_CLIENT_IP: &str = "-"; +const GRPC_STATUS: HeaderName = HeaderName::from_static("grpc-status"); +const X_REQUEST_ID: HeaderName = HeaderName::from_static("x-request-id"); + +#[derive(Debug)] +struct AccessLogContext { + method: Method, + route: String, + route_matched: bool, + client_ip: Option, + request_id: String, + started_at: Instant, + access_log: AccessLogSettings, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum AccessLogLevel { + Debug, + Info, + Warn, + Error, +} + +impl AccessLogContext { + fn from_http_request( + request: &Request, + server: &ApiServerSettings, + access_log: &AccessLogSettings, + ) -> Self { + if let Some(route) = request.extensions().get::() { + Self::from_request(request, server, access_log, route.as_str().to_string()) + } else { + Self::from_unmatched_http_request(request, server, access_log) + } + } + + fn from_grpc_request( + request: &Request, + server: &ApiServerSettings, + access_log: &AccessLogSettings, + ) -> Self { + Self::from_request( + request, + server, + access_log, + request.uri().path().to_string(), + ) + } + + fn from_request( + request: &Request, + server: &ApiServerSettings, + access_log: &AccessLogSettings, + route: String, + ) -> Self { + Self { + method: request.method().clone(), + route, + route_matched: true, + client_ip: effective_client_ip(request, server), + request_id: request_id(request.headers()), + started_at: Instant::now(), + access_log: access_log.clone(), + } + } + + fn from_unmatched_http_request( + request: &Request, + server: &ApiServerSettings, + access_log: &AccessLogSettings, + ) -> Self { + let mut context = Self::from_request( + request, + server, + access_log, + route_for_access_log(request.uri().path()), + ); + context.route_matched = false; + context + } + + fn client_ip_display(&self) -> String { + self.client_ip.map_or_else( + || UNKNOWN_CLIENT_IP.to_string(), + |client_ip| client_ip.to_string(), + ) + } + + fn log_http_completion( + &self, + status: StatusCode, + handler_latency: Duration, + response_bytes: u64, + completion: &'static str, + ) { + if !self.access_log.enabled { + return; + } + let client_ip = self.client_ip_display(); + let elapsed = self.started_at.elapsed(); + let latency_ms = duration_ms(elapsed); + let handler_latency_ms = duration_ms(handler_latency); + let slow = self.access_log.slow_request_threshold_ms != 0 + && handler_latency >= Duration::from_millis(self.access_log.slow_request_threshold_ms); + macro_rules! emit { + ($macro:ident) => { + tracing::$macro!( + target: ACCESS_LOG_TARGET, + protocol = "http", + method = %self.method, + route = %self.route, + route_matched = self.route_matched, + status = status.as_u16(), + latency_ms, + handler_latency_ms, + response_bytes, + slow, + client_ip = %client_ip, + request_id = %self.request_id, + completion, + "request completed" + ) + }; + } + match http_access_log_level(status, slow, completion) { + AccessLogLevel::Debug => emit!(debug), + AccessLogLevel::Info => emit!(info), + AccessLogLevel::Warn => emit!(warn), + AccessLogLevel::Error => emit!(error), + } + } + + fn log_grpc_completion( + &self, + http_status: StatusCode, + grpc_code: i32, + grpc_status: &str, + completion: &'static str, + ) { + let elapsed = self.started_at.elapsed(); + record_grpc_metrics(&self.route, grpc_code, grpc_status, elapsed); + if !self.access_log.enabled { + return; + } + let client_ip = self.client_ip_display(); + let latency_ms = duration_ms(elapsed); + macro_rules! emit { + ($macro:ident) => { + tracing::$macro!( + target: ACCESS_LOG_TARGET, + protocol = "grpc", + rpc = %self.route, + http_status = http_status.as_u16(), + grpc_status, + grpc_code, + latency_ms, + client_ip = %client_ip, + request_id = %self.request_id, + completion, + "request completed" + ) + }; + } + match grpc_access_log_level(http_status, grpc_code, completion) { + AccessLogLevel::Debug => emit!(debug), + AccessLogLevel::Info => emit!(info), + AccessLogLevel::Warn => emit!(warn), + AccessLogLevel::Error => emit!(error), + } + } +} + +/// Add request correlation and emit one completion log for an HTTP request. +pub async fn http_access_log_middleware( + request: Request, + next: Next, + server: &ApiServerSettings, + access_log: &AccessLogSettings, +) -> Response { + let context = AccessLogContext::from_http_request(&request, server, access_log); + + complete_http_request(request, next, context).await +} + +async fn complete_http_request( + request: Request, + next: Next, + context: AccessLogContext, +) -> Response { + let request_id = context.request_id.clone(); + let mut response = CURRENT_REQUEST_ID + .scope(request_id, async move { next.run(request).await }) + .await; + + insert_request_id_header(&mut response, &context.request_id); + if !context.access_log.enabled { + return response; + } + + let status = response.status(); + let handler_latency = context.started_at.elapsed(); + let expected_response_bytes = response + .headers() + .get(axum::http::header::CONTENT_LENGTH) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()); + response.map(|body| { + Body::new(HttpAccessLogBody::new( + body, + context, + status, + handler_latency, + expected_response_bytes, + )) + }) +} + +/// Add request correlation and emit a completion log with the final gRPC status. +/// Non-gRPC requests reaching the transport fallback are logged as unmatched HTTP. +pub async fn grpc_access_log_middleware( + request: Request, + next: Next, + server: &ApiServerSettings, + access_log: &AccessLogSettings, +) -> Response { + if !request_targets_grpc_transport(request.headers()).unwrap_or(false) { + let context = AccessLogContext::from_unmatched_http_request(&request, server, access_log); + return complete_http_request(request, next, context).await; + } + + let context = AccessLogContext::from_grpc_request(&request, server, access_log); + + let request_id = context.request_id.clone(); + let mut response = CURRENT_REQUEST_ID + .scope(request_id, async move { next.run(request).await }) + .await; + + insert_request_id_header(&mut response, &context.request_id); + let http_status = response.status(); + if let Some((grpc_code, grpc_status)) = grpc_status(response.headers()) { + context.log_grpc_completion(http_status, grpc_code, grpc_status, "finished"); + return response; + } + + if !http_status.is_success() { + context.log_grpc_completion(http_status, -1, "HTTP_ERROR", "finished"); + return response; + } + + response.map(|body| Body::new(GrpcAccessLogBody::new(body, context, http_status))) +} + +/// Return whether a request targets a gRPC or gRPC-Web transport. +pub fn request_targets_grpc_transport(headers: &HeaderMap) -> Result { + let Some(value) = headers.get(axum::http::header::CONTENT_TYPE) else { + return Ok(false); + }; + let media_type = value + .to_str()? + .trim() + .split(';') + .next() + .unwrap_or_default() + .trim() + .to_ascii_lowercase(); + Ok(media_type.starts_with("application/grpc")) +} + +#[derive(Debug)] +struct HttpAccessLogBody { + inner: Body, + context: Option, + status: StatusCode, + handler_latency: Duration, + expected_response_bytes: Option, + response_bytes: u64, +} + +impl HttpAccessLogBody { + fn new( + inner: Body, + context: AccessLogContext, + status: StatusCode, + handler_latency: Duration, + expected_response_bytes: Option, + ) -> Self { + let completes_without_body_poll = context.method == Method::HEAD + || status.is_informational() + || matches!(status, StatusCode::NO_CONTENT | StatusCode::NOT_MODIFIED) + || expected_response_bytes == Some(0); + let mut body = Self { + inner, + context: Some(context), + status, + handler_latency, + expected_response_bytes, + response_bytes: 0, + }; + if completes_without_body_poll || body.inner.is_end_stream() { + body.complete("finished"); + } + body + } + + fn complete(&mut self, completion: &'static str) { + if let Some(context) = self.context.take() { + context.log_http_completion( + self.status, + self.handler_latency, + self.response_bytes, + completion, + ); + } + } +} + +impl HttpBody for HttpAccessLogBody { + type Data = Bytes; + type Error = axum::Error; + + fn poll_frame( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, Self::Error>>> { + let this = self.get_mut(); + match Pin::new(&mut this.inner).poll_frame(cx) { + Poll::Ready(Some(Ok(frame))) => { + if let Some(data) = frame.data_ref() { + this.response_bytes = this + .response_bytes + .saturating_add(u64::try_from(data.len()).unwrap_or(u64::MAX)); + } + if this.inner.is_end_stream() + || this + .expected_response_bytes + .is_some_and(|expected| this.response_bytes >= expected) + { + this.complete("finished"); + } + Poll::Ready(Some(Ok(frame))) + } + Poll::Ready(Some(Err(error))) => { + this.complete("body_error"); + Poll::Ready(Some(Err(error))) + } + Poll::Ready(None) => { + this.complete("finished"); + Poll::Ready(None) + } + Poll::Pending => Poll::Pending, + } + } + + fn is_end_stream(&self) -> bool { + self.inner.is_end_stream() + } + + fn size_hint(&self) -> SizeHint { + self.inner.size_hint() + } +} + +impl Drop for HttpAccessLogBody { + fn drop(&mut self) { + self.complete("response_dropped"); + } +} + +#[derive(Debug)] +struct GrpcAccessLogBody { + inner: Body, + context: Option, + http_status: StatusCode, +} + +impl GrpcAccessLogBody { + fn new(inner: Body, context: AccessLogContext, http_status: StatusCode) -> Self { + let mut body = Self { + inner, + context: Some(context), + http_status, + }; + if body.inner.is_end_stream() { + body.complete(2, "UNKNOWN", "missing_status"); + } + body + } + + fn complete(&mut self, grpc_code: i32, grpc_status: &str, completion: &'static str) { + if let Some(context) = self.context.take() { + context.log_grpc_completion(self.http_status, grpc_code, grpc_status, completion); + } + } +} + +impl HttpBody for GrpcAccessLogBody { + type Data = Bytes; + type Error = axum::Error; + + fn poll_frame( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, Self::Error>>> { + let this = self.get_mut(); + match Pin::new(&mut this.inner).poll_frame(cx) { + Poll::Ready(Some(Ok(frame))) => { + if let Some(trailers) = frame.trailers_ref() { + if let Some((grpc_code, grpc_status)) = grpc_status(trailers) { + this.complete(grpc_code, grpc_status, "finished"); + } else { + this.complete(2, "UNKNOWN", "missing_status"); + } + } + Poll::Ready(Some(Ok(frame))) + } + Poll::Ready(Some(Err(error))) => { + this.complete(2, "UNKNOWN", "body_error"); + Poll::Ready(Some(Err(error))) + } + Poll::Ready(None) => { + this.complete(2, "UNKNOWN", "missing_status"); + Poll::Ready(None) + } + Poll::Pending => Poll::Pending, + } + } + + fn is_end_stream(&self) -> bool { + self.inner.is_end_stream() + } + + fn size_hint(&self) -> SizeHint { + self.inner.size_hint() + } +} + +impl Drop for GrpcAccessLogBody { + fn drop(&mut self) { + self.complete(1, "CANCELLED", "response_dropped"); + } +} + +fn request_id(headers: &HeaderMap) -> String { + headers + .get(&X_REQUEST_ID) + .and_then(|value| value.to_str().ok()) + .filter(|value| { + !value.is_empty() + && value.len() <= 64 + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_')) + }) + .map_or_else(|| synctv_common::snanoid!(12), str::to_owned) +} + +fn route_for_access_log(path: &str) -> String { + if path.len() <= MAX_LOGGED_ROUTE_LEN { + return path.to_string(); + } + + let mut end = MAX_LOGGED_ROUTE_LEN - 3; + while !path.is_char_boundary(end) { + end -= 1; + } + format!("{}...", &path[..end]) +} + +fn effective_client_ip(request: &Request, server: &ApiServerSettings) -> Option { + let peer_ip = request + .extensions() + .get::>() + .map(|connect_info| connect_info.0.ip())?; + Some( + synctv_adapter::client_ip::extract_client_ip_from_headers( + |ip| server.is_trusted_proxy(ip), + peer_ip, + request.headers(), + ) + .unwrap_or(peer_ip), + ) +} + +fn insert_request_id_header(response: &mut Response, request_id: &str) { + if let Ok(value) = HeaderValue::from_str(request_id) { + response.headers_mut().insert(X_REQUEST_ID, value); + } +} + +fn grpc_status(headers: &HeaderMap) -> Option<(i32, &'static str)> { + let code = headers + .get(&GRPC_STATUS)? + .to_str() + .ok()? + .parse::() + .ok()?; + Some((code, grpc_status_name(code))) +} + +fn http_access_log_level(status: StatusCode, slow: bool, completion: &str) -> AccessLogLevel { + if completion == "body_error" || status.is_server_error() { + AccessLogLevel::Error + } else if status.is_client_error() || slow { + AccessLogLevel::Warn + } else if completion == "response_dropped" { + AccessLogLevel::Debug + } else { + AccessLogLevel::Info + } +} + +fn grpc_access_log_level( + http_status: StatusCode, + grpc_code: i32, + completion: &str, +) -> AccessLogLevel { + if !http_status.is_success() || matches!(completion, "body_error" | "missing_status") { + return AccessLogLevel::Error; + } + match grpc_code { + 1 => AccessLogLevel::Debug, + 0 | 3 | 5 | 6 | 7 | 9 | 11 | 12 | 16 => AccessLogLevel::Info, + 4 | 8 | 10 | 14 => AccessLogLevel::Warn, + _ => AccessLogLevel::Error, + } +} + +fn record_grpc_metrics(route: &str, grpc_code: i32, grpc_status: &str, elapsed: Duration) { + let (service, method) = grpc_metric_labels(route, grpc_code); + metrics::REMOTE_TRANSPORT_REQUESTS_TOTAL + .with_label_values(&[service, method, grpc_status]) + .inc(); + metrics::REMOTE_TRANSPORT_REQUEST_DURATION + .with_label_values(&[service, method, grpc_status]) + .observe(elapsed.as_secs_f64()); +} + +fn grpc_metric_labels(route: &str, grpc_code: i32) -> (&str, &str) { + let mut segments = route.strip_prefix('/').unwrap_or(route).split('/'); + let service = segments.next().filter(|value| !value.is_empty()); + let method = segments.next().filter(|value| !value.is_empty()); + if segments.next().is_some() { + return ("", ""); + } + match (service, method) { + (Some(_), Some(_)) if grpc_code == 12 => ("", ""), + (Some(service), Some(method)) => (service, method), + _ => ("", ""), + } +} + +const fn grpc_status_name(code: i32) -> &'static str { + match code { + 0 => "OK", + 1 => "CANCELLED", + 2 => "UNKNOWN", + 3 => "INVALID_ARGUMENT", + 4 => "DEADLINE_EXCEEDED", + 5 => "NOT_FOUND", + 6 => "ALREADY_EXISTS", + 7 => "PERMISSION_DENIED", + 8 => "RESOURCE_EXHAUSTED", + 9 => "FAILED_PRECONDITION", + 10 => "ABORTED", + 11 => "OUT_OF_RANGE", + 12 => "UNIMPLEMENTED", + 13 => "INTERNAL", + 14 => "UNAVAILABLE", + 15 => "DATA_LOSS", + 16 => "UNAUTHENTICATED", + _ => "UNKNOWN_CODE", + } +} + +fn duration_ms(duration: Duration) -> f64 { + (duration.as_secs_f64() * 1_000_000.0).round() / 1_000.0 +} + +#[cfg(test)] +mod tests { + use super::*; + use std::{convert::Infallible, io::Write, sync::Arc}; + + use axum::{routing::get, Router}; + use futures::stream; + use http_body_util::{BodyExt, StreamBody}; + use tower::ServiceExt; + use tracing_subscriber::fmt::MakeWriter; + + type TestResult = Result<(), Box>; + + #[derive(Clone, Default)] + struct LogCapture(Arc>>); + + impl LogCapture { + fn contents(&self) -> String { + String::from_utf8( + self.0 + .lock() + .expect("log capture lock should not be poisoned") + .clone(), + ) + .expect("captured logs should be UTF-8") + } + } + + struct LogWriter(Arc>>); + + impl Write for LogWriter { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0 + .lock() + .expect("log capture lock should not be poisoned") + .extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl<'writer> MakeWriter<'writer> for LogCapture { + type Writer = LogWriter; + + fn make_writer(&'writer self) -> Self::Writer { + LogWriter(self.0.clone()) + } + } + + fn json_subscriber(capture: LogCapture) -> impl tracing::Subscriber + Send + Sync { + tracing_subscriber::fmt() + .json() + .with_ansi(false) + .with_writer(capture) + .finish() + } + + #[test] + fn request_id_accepts_safe_values_and_replaces_unsafe_values() -> TestResult { + let mut headers = HeaderMap::new(); + headers.insert(&X_REQUEST_ID, HeaderValue::from_static("request_ABC-123")); + assert_eq!(request_id(&headers), "request_ABC-123"); + + headers.insert(&X_REQUEST_ID, HeaderValue::from_static("contains spaces")); + let generated = request_id(&headers); + assert_eq!(generated.len(), 12); + assert!(generated.bytes().all(|byte| byte.is_ascii_alphanumeric())); + Ok(()) + } + + #[test] + fn client_ip_uses_forwarded_header_only_for_trusted_proxy() -> TestResult { + let peer = "192.0.2.10:8080".parse::()?; + let mut request = Request::builder() + .header("x-forwarded-for", "203.0.113.20") + .body(Body::empty())?; + request.extensions_mut().insert(ConnectInfo(peer)); + + let untrusted = ApiServerSettings::default(); + assert_eq!(effective_client_ip(&request, &untrusted), Some(peer.ip())); + + let mut trusted = ApiServerSettings::default(); + trusted.trusted_proxies.push(peer.ip().to_string()); + assert_eq!( + effective_client_ip(&request, &trusted), + Some("203.0.113.20".parse::()?) + ); + Ok(()) + } + + #[test] + fn grpc_status_names_cover_standard_and_unknown_codes() { + assert_eq!(grpc_status_name(0), "OK"); + assert_eq!(grpc_status_name(13), "INTERNAL"); + assert_eq!(grpc_status_name(99), "UNKNOWN_CODE"); + } + + #[test] + fn duration_ms_is_rounded_to_microsecond_precision() { + assert!((duration_ms(Duration::from_nanos(45_250)) - 0.045).abs() < f64::EPSILON); + assert!((duration_ms(Duration::from_micros(45)) - 0.045).abs() < f64::EPSILON); + } + + #[test] + fn access_log_levels_follow_transport_semantics() { + assert_eq!( + http_access_log_level(StatusCode::OK, false, "finished"), + AccessLogLevel::Info + ); + assert_eq!( + http_access_log_level(StatusCode::OK, true, "finished"), + AccessLogLevel::Warn + ); + assert_eq!( + http_access_log_level(StatusCode::OK, false, "response_dropped"), + AccessLogLevel::Debug + ); + assert_eq!( + http_access_log_level(StatusCode::OK, false, "body_error"), + AccessLogLevel::Error + ); + assert_eq!( + http_access_log_level(StatusCode::BAD_REQUEST, false, "finished"), + AccessLogLevel::Warn + ); + assert_eq!( + http_access_log_level(StatusCode::INTERNAL_SERVER_ERROR, false, "finished"), + AccessLogLevel::Error + ); + assert_eq!( + grpc_access_log_level(StatusCode::OK, 1, "response_dropped"), + AccessLogLevel::Debug + ); + assert_eq!( + grpc_access_log_level(StatusCode::OK, 3, "finished"), + AccessLogLevel::Info + ); + assert_eq!( + grpc_access_log_level(StatusCode::OK, 14, "finished"), + AccessLogLevel::Warn + ); + assert_eq!( + grpc_access_log_level(StatusCode::OK, 13, "finished"), + AccessLogLevel::Error + ); + } + + #[test] + fn grpc_metric_labels_bound_unimplemented_and_invalid_methods() { + assert_eq!( + grpc_metric_labels("/package.Service/Stream", 0), + ("package.Service", "Stream") + ); + assert_eq!( + grpc_metric_labels("/package.Service/attacker-controlled", 12), + ("", "") + ); + assert_eq!( + grpc_metric_labels("/attacker-controlled.Service/Method", 12), + ("", "") + ); + assert_eq!( + grpc_metric_labels("/invalid/too/many", 0), + ("", "") + ); + } + + #[test] + fn grpc_fallback_logs_unmatched_http_path_without_query() -> TestResult { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + let capture = LogCapture::default(); + let subscriber = json_subscriber(capture.clone()); + + tracing::subscriber::with_default(subscriber, || { + runtime.block_on(async { + let server = ApiServerSettings::default(); + let access_log = AccessLogSettings::default(); + let app = Router::new() + .route( + "/private-resource-id", + get(|| async { StatusCode::NOT_FOUND }), + ) + .layer(axum::middleware::from_fn(move |request, next| { + let server = server.clone(); + let access_log = access_log.clone(); + async move { + grpc_access_log_middleware(request, next, &server, &access_log).await + } + })); + let request = Request::builder() + .uri("/private-resource-id?secret=must-not-appear") + .header(&X_REQUEST_ID, "http-fallback-001") + .body(Body::empty())?; + + let response = app.oneshot(request).await?; + assert_eq!(response.status(), StatusCode::NOT_FOUND); + assert_eq!( + response.headers().get(&X_REQUEST_ID), + Some(&HeaderValue::from_static("http-fallback-001")) + ); + response.into_body().collect().await?; + TestResult::Ok(()) + }) + })?; + + let output = capture.contents(); + assert!(output.contains(r#""level":"WARN""#), "{output}"); + assert!(output.contains(r#""protocol":"http""#), "{output}"); + assert!(output.contains(r#""route":"/private-resource-id""#)); + assert!(output.contains(r#""route_matched":false"#)); + assert!(output.contains(r#""status":404"#)); + assert!(!output.contains(r#""protocol":"grpc""#)); + assert!(!output.contains("must-not-appear")); + Ok(()) + } + + #[test] + fn http_access_log_uses_matched_route_and_omits_query() -> TestResult { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + let capture = LogCapture::default(); + let subscriber = json_subscriber(capture.clone()); + + tracing::subscriber::with_default(subscriber, || { + runtime.block_on(async { + let server = ApiServerSettings::default(); + let access_log = AccessLogSettings::default(); + let app = Router::new() + .route( + "/items/{item_id}", + get(|| async { + let frames = stream::iter([Ok::<_, Infallible>(Frame::data( + Bytes::from_static(b"response-body"), + ))]); + Response::builder() + .status(StatusCode::INTERNAL_SERVER_ERROR) + .header(axum::http::header::CONTENT_LENGTH, "13") + .body(Body::new(StreamBody::new(frames))) + .expect("streaming response should build") + }), + ) + .layer(axum::middleware::from_fn(move |request, next| { + let server = server.clone(); + let access_log = access_log.clone(); + async move { + http_access_log_middleware(request, next, &server, &access_log).await + } + })); + let mut request = Request::builder() + .uri("/items/private-item?token=secret-value") + .header(&X_REQUEST_ID, "http-request-123") + .body(Body::empty())?; + request + .extensions_mut() + .insert(ConnectInfo("192.0.2.10:8080".parse::()?)); + + let response = app.oneshot(request).await?; + assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!( + response.headers().get(&X_REQUEST_ID), + Some(&HeaderValue::from_static("http-request-123")) + ); + assert!(capture.contents().is_empty()); + let mut body = response.into_body(); + let frame = body + .frame() + .await + .ok_or("response body should contain one frame")??; + assert_eq!( + frame.data_ref(), + Some(&Bytes::from_static(b"response-body")) + ); + assert!(!body.is_end_stream()); + drop(body); + TestResult::Ok(()) + }) + })?; + + let output = capture.contents(); + assert!(output.contains(r#""level":"ERROR""#), "{output}"); + assert!(output.contains(r#""protocol":"http""#)); + assert!(output.contains(r#""method":"GET""#)); + assert!(output.contains(r#""route":"/items/{item_id}""#)); + assert!(output.contains(r#""route_matched":true"#)); + assert!(output.contains(r#""status":500"#)); + assert!(output.contains(r#""response_bytes":13"#)); + assert!(output.contains(r#""slow":false"#)); + assert!(output.contains(r#""completion":"finished""#)); + assert!(output.contains(r#""client_ip":"192.0.2.10""#)); + assert!(output.contains(r#""request_id":"http-request-123""#)); + assert!(!output.contains("private-item")); + assert!(!output.contains("secret-value")); + Ok(()) + } + + #[test] + fn http_access_log_reports_body_errors_after_partial_delivery() -> TestResult { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + let capture = LogCapture::default(); + let subscriber = json_subscriber(capture.clone()); + + tracing::subscriber::with_default(subscriber, || { + runtime.block_on(async { + let server = ApiServerSettings::default(); + let access_log = AccessLogSettings::default(); + let app = Router::new() + .route( + "/stream", + get(|| async { + let frames = stream::iter([ + Ok::<_, std::io::Error>(Frame::data(Bytes::from_static( + b"partial", + ))), + Err(std::io::Error::other("stream failed")), + ]); + Response::builder() + .body(Body::new(StreamBody::new(frames))) + .expect("streaming response should build") + }), + ) + .layer(axum::middleware::from_fn(move |request, next| { + let server = server.clone(); + let access_log = access_log.clone(); + async move { + http_access_log_middleware(request, next, &server, &access_log).await + } + })); + let request = Request::builder() + .uri("/stream") + .header(&X_REQUEST_ID, "http-stream-error") + .body(Body::empty())?; + + let response = app.oneshot(request).await?; + assert!(capture.contents().is_empty()); + assert!(response.into_body().collect().await.is_err()); + TestResult::Ok(()) + }) + })?; + + let output = capture.contents(); + assert!(output.contains(r#""level":"ERROR""#), "{output}"); + assert!(output.contains(r#""response_bytes":7"#)); + assert!(output.contains(r#""completion":"body_error""#)); + assert!(output.contains(r#""request_id":"http-stream-error""#)); + Ok(()) + } + + #[test] + fn http_access_log_completes_responses_that_have_no_wire_body() -> TestResult { + let capture = LogCapture::default(); + let subscriber = json_subscriber(capture.clone()); + + tracing::subscriber::with_default(subscriber, || { + for (method, status, expected_response_bytes) in [ + (Method::HEAD, StatusCode::OK, Some(13)), + (Method::GET, StatusCode::NO_CONTENT, None), + (Method::GET, StatusCode::OK, Some(0)), + ] { + let request = Request::builder() + .method(method) + .uri("/resource") + .body(Body::empty())?; + let context = AccessLogContext::from_request( + &request, + &ApiServerSettings::default(), + &AccessLogSettings::default(), + "/resource".to_string(), + ); + let frames = stream::pending::, Infallible>>(); + let body = HttpAccessLogBody::new( + Body::new(StreamBody::new(frames)), + context, + status, + Duration::ZERO, + expected_response_bytes, + ); + drop(body); + } + TestResult::Ok(()) + })?; + + let output = capture.contents(); + assert_eq!(output.matches(r#""completion":"finished""#).count(), 3); + assert_eq!(output.matches(r#""response_bytes":0"#).count(), 3); + assert!(!output.contains("response_dropped")); + Ok(()) + } + + #[test] + fn disabled_access_log_still_returns_request_id() -> TestResult { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + let capture = LogCapture::default(); + let subscriber = json_subscriber(capture.clone()); + + tracing::subscriber::with_default(subscriber, || { + runtime.block_on(async { + let server = ApiServerSettings::default(); + let access_log = AccessLogSettings { + enabled: false, + ..AccessLogSettings::default() + }; + let app = Router::new() + .route("/health", get(|| async { StatusCode::OK })) + .layer(axum::middleware::from_fn(move |request, next| { + let server = server.clone(); + let access_log = access_log.clone(); + async move { + http_access_log_middleware(request, next, &server, &access_log).await + } + })); + let request = Request::builder() + .uri("/health") + .header(&X_REQUEST_ID, "request-without-log") + .body(Body::empty())?; + + let response = app.oneshot(request).await?; + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + response.headers().get(&X_REQUEST_ID), + Some(&HeaderValue::from_static("request-without-log")) + ); + TestResult::Ok(()) + }) + })?; + + assert!(capture.contents().is_empty()); + Ok(()) + } + + #[test] + fn grpc_access_log_waits_for_trailers_and_reports_final_status() -> TestResult { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + let capture = LogCapture::default(); + let subscriber = json_subscriber(capture.clone()); + + tracing::subscriber::with_default(subscriber, || { + runtime.block_on(async { + let server = ApiServerSettings::default(); + let access_log = AccessLogSettings::default(); + let app = Router::new() + .route( + "/package.Service/{method}", + axum::routing::post(|| async { + let mut trailers = HeaderMap::new(); + trailers.insert(GRPC_STATUS, HeaderValue::from_static("13")); + let frames = + stream::iter([Ok::<_, Infallible>(Frame::trailers(trailers))]); + Response::builder() + .header(axum::http::header::CONTENT_TYPE, "application/grpc") + .body(Body::new(StreamBody::new(frames))) + .expect("gRPC test response should build") + }), + ) + .layer(axum::middleware::from_fn(move |request, next| { + let server = server.clone(); + let access_log = access_log.clone(); + async move { + grpc_access_log_middleware(request, next, &server, &access_log).await + } + })); + let request = Request::builder() + .method(Method::POST) + .uri("/package.Service/Stream?authorization=secret-value") + .header(axum::http::header::CONTENT_TYPE, "application/grpc") + .header(&X_REQUEST_ID, "grpc-request-123") + .body(Body::empty())?; + + let response = app.oneshot(request).await?; + assert!(capture.contents().is_empty()); + assert_eq!( + response.headers().get(&X_REQUEST_ID), + Some(&HeaderValue::from_static("grpc-request-123")) + ); + response.into_body().collect().await?; + TestResult::Ok(()) + }) + })?; + + let output = capture.contents(); + assert!(output.contains(r#""level":"ERROR""#), "{output}"); + assert!(output.contains(r#""protocol":"grpc""#)); + assert!(output.contains(r#""rpc":"/package.Service/Stream""#)); + assert!(output.contains(r#""grpc_status":"INTERNAL""#)); + assert!(output.contains(r#""grpc_code":13"#)); + assert!(output.contains(r#""completion":"finished""#)); + assert!(output.contains(r#""request_id":"grpc-request-123""#)); + assert!(!output.contains("authorization")); + assert!(!output.contains("secret-value")); + Ok(()) + } +} diff --git a/synctv-api-grpc/src/grpc/mod.rs b/synctv-api-grpc/src/grpc/mod.rs index bac7078e..cb35acb0 100644 --- a/synctv-api-grpc/src/grpc/mod.rs +++ b/synctv-api-grpc/src/grpc/mod.rs @@ -293,23 +293,6 @@ const fn grpc_service_registration_plan( } } -fn request_targets_grpc_transport( - headers: &axum::http::HeaderMap, -) -> Result { - let Some(value) = headers.get(axum::http::header::CONTENT_TYPE) else { - return Ok(false); - }; - let media_type = value - .to_str()? - .trim() - .split(';') - .next() - .unwrap_or_default() - .trim() - .to_ascii_lowercase(); - Ok(media_type.starts_with("application/grpc")) -} - fn relay_cancellation_token( mut shutdown_rx: Option>, ) -> tokio_util::sync::CancellationToken { @@ -332,7 +315,8 @@ async fn grpc_transport_only_middleware( request: axum::extract::Request, next: axum::middleware::Next, ) -> axum::response::Response { - match request_targets_grpc_transport(request.headers()) { + match synctv_api_common::transport_access_log::request_targets_grpc_transport(request.headers()) + { Ok(true) => next.run(request).await, Ok(false) => axum::response::IntoResponse::into_response(axum::http::StatusCode::NOT_FOUND), Err(_) => axum::response::IntoResponse::into_response(axum::http::StatusCode::BAD_REQUEST), @@ -2225,10 +2209,25 @@ async fn build_axum_router_with_health( tracing::info!("gRPC reflection service registered"); } + let access_log_server_config = Arc::new(runtime_settings.server.clone()); + let access_log_config = Arc::new(runtime_settings.access_log.clone()); let router = routes .routes() .into_axum_router() - .layer(axum::middleware::from_fn(grpc_transport_only_middleware)); + .layer(axum::middleware::from_fn(grpc_transport_only_middleware)) + .layer(axum::middleware::from_fn(move |request, next| { + let server_config = Arc::clone(&access_log_server_config); + let access_log = Arc::clone(&access_log_config); + async move { + synctv_api_common::transport_access_log::grpc_access_log_middleware( + request, + next, + server_config.as_ref(), + access_log.as_ref(), + ) + .await + } + })); Ok(BuiltGrpcRouter { router, @@ -2657,7 +2656,7 @@ mod tests { fn test_request_targets_grpc_transport_requires_grpc_content_type() -> TestResult { let mut headers = axum::http::HeaderMap::new(); assert!( - !super::request_targets_grpc_transport(&headers)?, + !synctv_api_common::transport_access_log::request_targets_grpc_transport(&headers)?, "requests without Content-Type must not be treated as gRPC" ); @@ -2666,7 +2665,7 @@ mod tests { axum::http::HeaderValue::from_static("application/json"), ); assert!( - !super::request_targets_grpc_transport(&headers)?, + !synctv_api_common::transport_access_log::request_targets_grpc_transport(&headers)?, "plain HTTP JSON requests must not be treated as gRPC" ); @@ -2675,7 +2674,7 @@ mod tests { axum::http::HeaderValue::from_static("application/grpc"), ); assert!( - super::request_targets_grpc_transport(&headers)?, + synctv_api_common::transport_access_log::request_targets_grpc_transport(&headers)?, "canonical gRPC requests must be routed to tonic" ); @@ -2684,7 +2683,7 @@ mod tests { axum::http::HeaderValue::from_static("application/grpc+proto; charset=utf-8"), ); assert!( - super::request_targets_grpc_transport(&headers)?, + synctv_api_common::transport_access_log::request_targets_grpc_transport(&headers)?, "gRPC content-type variants must still be routed to tonic" ); Ok(()) @@ -2699,7 +2698,8 @@ mod tests { ); assert!( - super::request_targets_grpc_transport(&headers).is_err(), + synctv_api_common::transport_access_log::request_targets_grpc_transport(&headers) + .is_err(), "invalid Content-Type bytes must not be silently treated as a non-gRPC request" ); Ok(()) diff --git a/synctv-api-grpc/src/grpc_support.rs b/synctv-api-grpc/src/grpc_support.rs index 9861fd4d..df799c47 100644 --- a/synctv-api-grpc/src/grpc_support.rs +++ b/synctv-api-grpc/src/grpc_support.rs @@ -10,7 +10,12 @@ #[must_use] pub fn map_api_error_ref(err: &synctv_api_common::impls::ApiError) -> tonic::Status { let sanitized = synctv_api_common::api_error_model::sanitized_api_error(err); - synctv_api_common::api_error_model::GoogleApiError::from_api_error(&sanitized).to_tonic_status() + let request_id = synctv_api_common::request_context::CURRENT_REQUEST_ID + .try_with(Clone::clone) + .ok(); + synctv_api_common::api_error_model::GoogleApiError::from_api_error(&sanitized) + .with_request_id(request_id.as_deref()) + .to_tonic_status() } pub fn map_api_error(err: impl Into) -> tonic::Status { @@ -99,6 +104,7 @@ pub const fn grpc_unary_request_timeout() -> std::time::Duration { #[cfg(test)] mod tests { use super::request_metadata; + use tonic_types::StatusExt; type TestResult = Result<(), Box>; @@ -161,4 +167,21 @@ mod tests { assert!(metadata.user_agent.is_none()); Ok(()) } + + #[tokio::test] + async fn mapped_api_error_includes_current_request_id() -> TestResult { + let status = synctv_api_common::request_context::CURRENT_REQUEST_ID + .scope("grpc-request-123".to_string(), async { + super::map_api_error(synctv_api_common::impls::ApiError::InvalidInput( + "invalid request".to_string(), + )) + }) + .await; + + let request_info = status + .get_details_request_info() + .ok_or("missing RequestInfo error detail")?; + assert_eq!(request_info.request_id, "grpc-request-123"); + Ok(()) + } } diff --git a/synctv-api-http/src/http/metrics_middleware.rs b/synctv-api-http/src/http/metrics_middleware.rs index 1d0154b4..5c997835 100644 --- a/synctv-api-http/src/http/metrics_middleware.rs +++ b/synctv-api-http/src/http/metrics_middleware.rs @@ -1,6 +1,10 @@ //! Axum middleware for collecting HTTP request metrics. -use axum::{extract::Request, middleware::Next, response::Response}; +use axum::{ + extract::{MatchedPath, Request}, + middleware::Next, + response::Response, +}; use std::time::Instant; use synctv_api_common::observability::metrics; @@ -23,7 +27,10 @@ impl Drop for InFlightRequestGuard { /// Middleware that records HTTP request count, duration, and in-flight gauge. pub async fn metrics_layer(request: Request, next: Next) -> Response { let method = request.method().to_string(); - let path = metrics::normalize_path(request.uri().path()); + let path = request.extensions().get::().map_or_else( + || "".to_string(), + |path| path.as_str().to_string(), + ); let _in_flight = InFlightRequestGuard::new(); let start = Instant::now(); @@ -42,3 +49,38 @@ pub async fn metrics_layer(request: Request, next: Next) -> Response { response } + +#[cfg(test)] +mod tests { + use axum::{ + body::Body, + http::{Request, StatusCode}, + routing::get, + Router, + }; + use tower::ServiceExt; + + #[tokio::test] + async fn metrics_use_matched_route_without_resource_ids() { + let app = Router::new() + .route("/items/{item_id}", get(|| async { StatusCode::OK })) + .layer(axum::middleware::from_fn(super::metrics_layer)); + + let response = app + .oneshot( + Request::builder() + .uri("/items/private-resource-id") + .body(Body::empty()) + .expect("request should build"), + ) + .await + .expect("request should complete"); + assert_eq!(response.status(), StatusCode::OK); + + let output = synctv_api_common::observability::metrics::gather_metrics(); + assert!(output.contains( + "http_requests_total{method=\"GET\",path=\"/items/{item_id}\",status=\"200\"}" + )); + assert!(!output.contains("private-resource-id")); + } +} diff --git a/synctv-api-http/src/http/middleware.rs b/synctv-api-http/src/http/middleware.rs index 9da1bd70..50befc62 100644 --- a/synctv-api-http/src/http/middleware.rs +++ b/synctv-api-http/src/http/middleware.rs @@ -11,8 +11,6 @@ use std::sync::LazyLock; use super::{optional_header_str, AppError, AppState}; -pub use synctv_api_common::request_context::CURRENT_REQUEST_ID; - /// Transport metadata extracted from the HTTP request without performing /// authentication, blacklist, rate-limit, or timeout decisions. #[derive(Debug, Clone)] @@ -75,55 +73,6 @@ where } } -/// HTTP header name for request/trace ID propagation. -static X_REQUEST_ID: LazyLock = - LazyLock::new(|| axum::http::HeaderName::from_static("x-request-id")); - -/// Middleware that generates a unique request ID per request. -/// -/// - If the client sends an `X-Request-ID` header whose value is a non-empty -/// alphanumeric ASCII string of at most 64 characters, that value is reused -/// (allows end-to-end trace correlation from trusted clients). -/// - Otherwise a fresh 12-character shared base62 request ID is generated. -/// -/// The request ID is: -/// 1. Recorded in the current tracing span as `request_id` for log correlation. -/// 2. Echoed back in the `X-Request-ID` response header so callers can correlate -/// logs with their own request tracking. -/// 3. Exposed via a task-local so `AppError` responses can include it without -/// buffering and rewriting response bodies. -pub async fn request_id_middleware(request: Request, next: Next) -> Response { - // Honour an incoming X-Request-ID header when safe to do so. - let request_id = request - .headers() - .get(X_REQUEST_ID.clone()) - .and_then(|v| v.to_str().ok()) - .filter(|s| { - // Validate: non-empty, max 64 chars, alphanumeric + hyphens/underscores only. - let len = s.len(); - len > 0 - && len <= 64 - && s.bytes() - .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_') - }) - .map_or_else(|| synctv_common::snanoid!(12), str::to_owned); - - // Record in current tracing span for log correlation. - tracing::Span::current().record("requestId", request_id.as_str()); - tracing::debug!(request_id = %request_id, "Request received"); - - let mut response = CURRENT_REQUEST_ID - .scope(request_id.clone(), async move { next.run(request).await }) - .await; - - // Echo back in response header so callers can correlate. - if let Ok(value) = axum::http::HeaderValue::from_str(&request_id) { - response.headers_mut().insert(X_REQUEST_ID.clone(), value); - } - - response -} - /// Pre-validated security header names (validated once at startup via Lazy) static X_FRAME_OPTIONS: LazyLock = LazyLock::new(|| axum::http::HeaderName::from_static("x-frame-options")); diff --git a/synctv-api-http/src/http/mod.rs b/synctv-api-http/src/http/mod.rs index 729d6750..41ad7f15 100644 --- a/synctv-api-http/src/http/mod.rs +++ b/synctv-api-http/src/http/mod.rs @@ -49,9 +49,7 @@ use tower_http::compression::{ }; use tower_http::cors::CorsLayer; use tower_http::on_early_drop::{EarlyDropsAsFailures, OnEarlyDropLayer}; -use tower_http::trace::{ - DefaultMakeSpan, DefaultOnFailure, DefaultOnRequest, DefaultOnResponse, TraceLayer, -}; +use tower_http::trace::DefaultOnFailure; pub use auth::extract_client_ip; pub use error::{map_api_error, AppError, AppResult}; @@ -1791,8 +1789,8 @@ fn should_compress_application_response( }) } -/// Apply shared transport layers (CORS, body limit, security headers, HSTS, -/// request ID propagation, and tracing) and bind state. +/// Apply shared transport layers (CORS, compression, body limit, security +/// headers, and HSTS) and bind state. fn apply_shared_http_layers( router: Router, cors: CorsLayer, @@ -1811,7 +1809,6 @@ fn apply_shared_http_layers( ), ) .layer(axum::extract::DefaultBodyLimit::max(10 * 1024 * 1024)) - .layer(axum_middleware::from_fn(middleware::request_id_middleware)) .layer(axum_middleware::from_fn( middleware::security_headers_middleware, )) @@ -1854,6 +1851,8 @@ fn apply_shared_http_layers( fn apply_global_layers(router: Router, state: &AppState) -> anyhow::Result { let cors = build_cors_layer(&state.runtime_settings)?; let server_config = state.runtime_settings.server.clone(); + let access_log_server_config = Arc::new(server_config.clone()); + let access_log_config = Arc::new(state.runtime_settings.access_log.clone()); let hsts_value = middleware::hsts_header(63_072_000, true, false); Ok( apply_shared_http_layers(router, cors, server_config, hsts_value) @@ -1861,12 +1860,19 @@ fn apply_global_layers(router: Router, state: &AppState) -> anyhow::Re .layer(OnEarlyDropLayer::new(EarlyDropsAsFailures::new( DefaultOnFailure::default(), ))) - .layer( - TraceLayer::new_for_http() - .make_span_with(DefaultMakeSpan::new().level(tracing::Level::DEBUG)) - .on_request(DefaultOnRequest::new().level(tracing::Level::DEBUG)) - .on_response(DefaultOnResponse::new().level(tracing::Level::INFO)), - ) + .layer(axum_middleware::from_fn(move |request, next| { + let server_config = Arc::clone(&access_log_server_config); + let access_log = Arc::clone(&access_log_config); + async move { + synctv_api_common::transport_access_log::http_access_log_middleware( + request, + next, + server_config.as_ref(), + access_log.as_ref(), + ) + .await + } + })) .with_state(state.clone()), ) } diff --git a/synctv-core/Cargo.toml b/synctv-core/Cargo.toml index 02e8b371..a4c81065 100644 --- a/synctv-core/Cargo.toml +++ b/synctv-core/Cargo.toml @@ -84,6 +84,7 @@ failsafe.workspace = true # Time chrono.workspace = true +chrono-tz.workspace = true rsntp.workspace = true # Logging diff --git a/synctv-core/src/logging.rs b/synctv-core/src/logging.rs index 331db900..2ebfd4c4 100644 --- a/synctv-core/src/logging.rs +++ b/synctv-core/src/logging.rs @@ -4,12 +4,14 @@ use std::{ sync::OnceLock, }; +use chrono::{DateTime, Utc}; +use chrono_tz::Tz; use tracing::Level; use tracing_appender::non_blocking::{ErrorCounter, NonBlocking, NonBlockingBuilder, WorkerGuard}; use tracing_appender::rolling::{RollingFileAppender, Rotation}; use tracing_subscriber::{ filter::FilterFn, - fmt::{self, format::FmtSpan, writer::BoxMakeWriter}, + fmt::{self, format::Writer, time::FormatTime, writer::BoxMakeWriter}, layer::{Layer, Layered, SubscriberExt}, registry::Registry, util::SubscriberInitExt, @@ -17,6 +19,7 @@ use tracing_subscriber::{ const SQLX_POSTGRES_NOTICE_TARGET: &str = "sqlx::postgres::notice"; const LOG_BUFFERED_LINES_LIMIT: usize = 128_000; +const LOG_TIMESTAMP_FORMAT: &str = "%Y-%m-%dT%H:%M:%S%.6f%:z"; static LOGGING_ERROR_COUNTERS: OnceLock> = OnceLock::new(); #[derive(Debug, Clone)] @@ -39,6 +42,12 @@ pub enum LogRotation { Never, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LogStyle { + Diagnostic, + Access, +} + #[derive(Debug, Clone, PartialEq, Eq)] pub enum LogOutput { Stdout, @@ -53,6 +62,7 @@ pub enum LogOutput { #[derive(Debug, Clone)] pub struct ComponentLoggingOptions { pub name: String, + pub style: LogStyle, pub targets: Vec, pub level: String, pub format: String, @@ -62,6 +72,7 @@ pub struct ComponentLoggingOptions { #[derive(Debug, Clone)] pub struct LoggingOptions { + pub timezone: String, pub global: ComponentLoggingOptions, pub components: Vec, } @@ -69,8 +80,10 @@ pub struct LoggingOptions { impl Default for LoggingOptions { fn default() -> Self { Self { + timezone: "UTC".to_string(), global: ComponentLoggingOptions { name: "global".to_string(), + style: LogStyle::Diagnostic, targets: Vec::new(), level: "info".to_string(), format: "text".to_string(), @@ -82,6 +95,39 @@ impl Default for LoggingOptions { } } +#[derive(Debug, Clone, Copy)] +struct ConfiguredTimezoneTimer { + timezone: Tz, +} + +impl FormatTime for ConfiguredTimezoneTimer { + fn format_time(&self, writer: &mut Writer<'_>) -> std::fmt::Result { + write_log_timestamp(writer, self.timezone, Utc::now()) + } +} + +fn write_log_timestamp( + writer: &mut impl std::fmt::Write, + timezone: Tz, + timestamp: DateTime, +) -> std::fmt::Result { + write!( + writer, + "{}", + timestamp + .with_timezone(&timezone) + .format(LOG_TIMESTAMP_FORMAT) + ) +} + +#[cfg(test)] +fn format_log_timestamp(timezone: Tz, timestamp: DateTime) -> String { + let mut output = String::new(); + write_log_timestamp(&mut output, timezone, timestamp) + .expect("writing a timestamp to a string should succeed"); + output +} + /// Keeps every non-blocking logging worker alive until application shutdown. #[derive(Debug)] pub struct LoggingGuards { @@ -136,6 +182,11 @@ type LoggingSubscriber = Layered + Send + Sync>>, Re fn build_subscriber(config: &LoggingOptions) -> anyhow::Result<(LoggingSubscriber, LoggingGuards)> { validate_component_routes(config)?; + let timer = ConfiguredTimezoneTimer { + timezone: synctv_common::time::parse_timezone_name(&config.timezone).map_err(|error| { + anyhow::anyhow!("invalid logging timezone '{}': {error}", config.timezone) + })?, + }; let component_targets: Vec = config .components @@ -175,10 +226,24 @@ fn build_subscriber(config: &LoggingOptions) -> anyhow::Result<(LoggingSubscribe }); let writer = writers.writer_for(component)?; - let layer = if component.format.eq_ignore_ascii_case("json") { + let is_access_log = component.style == LogStyle::Access; + let layer = if component.format.eq_ignore_ascii_case("json") && is_access_log { fmt::layer() + .with_timer(timer) + .json() + .with_current_span(false) + .with_span_list(false) + .with_target(false) + .with_line_number(false) + .with_file(false) + .with_ansi(false) + .with_writer(writer) + .with_filter(filter) + .boxed() + } else if component.format.eq_ignore_ascii_case("json") { + fmt::layer() + .with_timer(timer) .json() - .with_span_events(FmtSpan::CLOSE) .with_current_span(true) .with_span_list(true) .with_target(true) @@ -188,10 +253,21 @@ fn build_subscriber(config: &LoggingOptions) -> anyhow::Result<(LoggingSubscribe .with_writer(writer) .with_filter(filter) .boxed() + } else if component.format.eq_ignore_ascii_case("text") && is_access_log { + fmt::layer() + .with_timer(timer) + .compact() + .with_target(false) + .with_line_number(false) + .with_file(false) + .with_ansi(ansi_enabled(component)) + .with_writer(writer) + .with_filter(filter) + .boxed() } else if component.format.eq_ignore_ascii_case("text") { fmt::layer() + .with_timer(timer) .compact() - .with_span_events(FmtSpan::CLOSE) .with_target(true) .with_line_number(true) .with_file(false) @@ -463,14 +539,54 @@ mod tests { #[test] fn default_logging_has_a_global_output() { let config = LoggingOptions::default(); + assert_eq!(config.timezone, "UTC"); assert_eq!(config.global.name, "global"); assert!(config.global.targets.is_empty()); assert!(config.components.is_empty()); } + #[test] + fn log_timestamp_uses_configured_timezone_and_daylight_saving_offset() { + let timezone = synctv_common::time::parse_timezone_name("America/New_York") + .expect("timezone should be valid"); + let winter = DateTime::parse_from_rfc3339("2026-01-15T12:00:00Z") + .expect("timestamp should be valid") + .to_utc(); + let summer = DateTime::parse_from_rfc3339("2026-07-15T12:00:00Z") + .expect("timestamp should be valid") + .to_utc(); + + assert_eq!( + format_log_timestamp(timezone, winter), + "2026-01-15T07:00:00.000000-05:00" + ); + assert_eq!( + format_log_timestamp(timezone, summer), + "2026-07-15T08:00:00.000000-04:00" + ); + assert_eq!( + format_log_timestamp(chrono_tz::UTC, summer), + "2026-07-15T12:00:00.000000+00:00" + ); + } + + #[test] + fn invalid_logging_timezone_is_rejected() { + let config = LoggingOptions { + timezone: "Invalid/Timezone".to_string(), + ..LoggingOptions::default() + }; + + let Err(error) = build_subscriber(&config) else { + panic!("invalid timezone should be rejected"); + }; + assert!(error.to_string().contains("invalid logging timezone")); + } + #[test] fn effective_level_comes_from_global_output() { let config = LoggingOptions { + timezone: "UTC".to_string(), global: ComponentLoggingOptions { level: "debug".to_string(), ..default_component("global") @@ -514,6 +630,7 @@ mod tests { #[test] fn overlapping_specialized_routes_are_rejected() { let config = LoggingOptions { + timezone: "UTC".to_string(), global: default_component("global"), components: vec![ ComponentLoggingOptions { @@ -532,6 +649,7 @@ mod tests { #[test] fn components_sharing_standard_output_have_independent_workers() { let config = LoggingOptions { + timezone: "UTC".to_string(), global: default_component("global"), components: vec![ComponentLoggingOptions { targets: vec!["synctv::health".to_string()], @@ -574,6 +692,7 @@ mod tests { #[test] fn zero_file_retention_is_rejected() { let config = LoggingOptions { + timezone: "UTC".to_string(), global: ComponentLoggingOptions { output: LogOutput::File { path: PathBuf::from("global.log"), @@ -621,6 +740,7 @@ mod tests { fn component_layers_use_independent_routes_levels_formats_and_files() { let dir = tempdir().expect("temporary log directory should be created"); let config = LoggingOptions { + timezone: "Asia/Shanghai".to_string(), global: ComponentLoggingOptions { format: "json".to_string(), output: LogOutput::File { @@ -649,21 +769,68 @@ mod tests { tracing::info!(target: "synctv_core::cache", "global-only-event"); tracing::info!(target: "synctv::health", "filtered-health-event"); tracing::warn!(target: "synctv::health", "health-only-event"); + let span = tracing::info_span!("silent-span"); + drop(span.enter()); }); drop(guards); let global_log = read_log_with_prefix(dir.path(), "global"); let health_log = read_log_with_prefix(dir.path(), "health"); assert!(global_log.contains("global-only-event")); + assert!(global_log.contains("+08:00")); assert!(global_log.contains("\"target\":\"synctv_core::cache\"")); assert!(!global_log.contains("health-only-event")); assert!(!global_log.contains("filtered-health-event")); + assert!(!global_log.contains("silent-span")); assert!(health_log.contains("health-only-event")); + assert!(health_log.contains("+08:00")); assert!(!health_log.contains("filtered-health-event")); assert!(!health_log.contains("global-only-event")); assert!(!health_log.trim_start().starts_with('{')); } + #[test] + fn access_component_uses_compact_context_free_text() { + let dir = tempdir().expect("temporary log directory should be created"); + let config = LoggingOptions { + timezone: "UTC".to_string(), + global: default_component("global"), + components: vec![ComponentLoggingOptions { + name: "access".to_string(), + style: LogStyle::Access, + targets: vec!["synctv::access".to_string()], + output: LogOutput::File { + path: dir.path().join("access.log"), + rotation: LogRotation::Never, + max_files: 2, + }, + ..default_component("access") + }], + }; + let (subscriber, guards) = + build_subscriber(&config).expect("logging subscriber should build"); + + tracing::subscriber::with_default(subscriber, || { + let span = tracing::info_span!("request", secret = "must-not-be-inherited"); + let _guard = span.enter(); + tracing::info!( + target: "synctv::access", + protocol = "http", + status = 200, + "request completed" + ); + }); + drop(guards); + + let access_log = read_log_with_prefix(dir.path(), "access"); + assert!(access_log.contains("request completed")); + assert!(access_log.contains("protocol=\"http\"")); + assert!(access_log.contains("status=200")); + assert!(!access_log.contains("synctv::access")); + assert!(!access_log.contains("must-not-be-inherited")); + assert!(!access_log.contains("logging.rs")); + } + fn read_log_with_prefix(dir: &Path, prefix: &str) -> String { let path = fs::read_dir(dir) .expect("log directory should be readable") @@ -684,6 +851,7 @@ mod tests { fn default_component(name: &str) -> ComponentLoggingOptions { ComponentLoggingOptions { name: name.to_string(), + style: LogStyle::Diagnostic, targets: Vec::new(), level: "info".to_string(), format: "text".to_string(), diff --git a/synctv.example.yaml b/synctv.example.yaml index 6c75a364..eb7b2f73 100644 --- a/synctv.example.yaml +++ b/synctv.example.yaml @@ -86,6 +86,22 @@ server: # rotation: # strategy: "daily" # max_files: 30 + # One completion line per public HTTP or gRPC request. This output is routed + # independently from server diagnostics. It can reuse a validated X-Request-ID, + # and never includes other headers, query strings, metadata, cookies, or bodies. + # Environment variables: SYNCTV_SERVER_ACCESS_LOG_ENABLED, + # SYNCTV_SERVER_ACCESS_LOG_SLOW_REQUEST_THRESHOLD_MS, + # SYNCTV_SERVER_ACCESS_LOG_LEVEL, SYNCTV_SERVER_ACCESS_LOG_FORMAT, + # SYNCTV_SERVER_ACCESS_LOG_COLOR, SYNCTV_SERVER_ACCESS_LOG_OUTPUT, and the + # standard *_OUTPUT_PATH / *_OUTPUT_ROTATION_* variants. + access_log: + enabled: true + # Set to 0 to disable slow HTTP request classification. + slow_request_threshold_ms: 1000 + level: "info" + format: "text" + output: "stdout" + color: "auto" time: # Default IANA timezone used for human-readable time output and local datetime parsing. diff --git a/synctv/src/app.rs b/synctv/src/app.rs index 4311616f..c5889638 100644 --- a/synctv/src/app.rs +++ b/synctv/src/app.rs @@ -2067,6 +2067,7 @@ mod tests { advertise_host: String::new(), shutdown_drain_timeout_seconds: 30, logging: crate::app_config::LoggingConfig::default(), + access_log: crate::app_config::AccessLogConfig::default(), }, health: crate::app_config::HealthConfig::default(), time: crate::app_config::TimeConfig::default(), diff --git a/synctv/src/app_config/mod.rs b/synctv/src/app_config/mod.rs index 1d6a0e22..eb8b5a51 100644 --- a/synctv/src/app_config/mod.rs +++ b/synctv/src/app_config/mod.rs @@ -100,6 +100,7 @@ pub struct ServerConfig { pub grpc_max_message_size_bytes: usize, pub grpc_compression_enabled: bool, pub logging: LoggingConfig, + pub access_log: AccessLogConfig, } impl Default for ServerConfig { @@ -116,6 +117,26 @@ impl Default for ServerConfig { grpc_max_message_size_bytes: 16 * 1024 * 1024, grpc_compression_enabled: true, logging: LoggingConfig::default(), + access_log: AccessLogConfig::default(), + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(default)] +pub struct AccessLogConfig { + pub enabled: bool, + pub slow_request_threshold_ms: u64, + #[serde(flatten)] + pub logging: LoggingConfig, +} + +impl Default for AccessLogConfig { + fn default() -> Self { + Self { + enabled: true, + slow_request_threshold_ms: 1_000, + logging: LoggingConfig::default(), } } } diff --git a/synctv/src/app_config/validation.rs b/synctv/src/app_config/validation.rs index 2fbcd937..5fc3b571 100644 --- a/synctv/src/app_config/validation.rs +++ b/synctv/src/app_config/validation.rs @@ -612,6 +612,7 @@ impl AppConfig { for (path, logging) in [ ("logging", &self.logging), ("server.logging", &self.server.logging), + ("server.access_log", &self.server.access_log.logging), ("health.logging", &self.health.logging), ("metrics.logging", &self.metrics.logging), ("cluster.logging", &self.cluster.logging), diff --git a/synctv/src/bootstrap/database.rs b/synctv/src/bootstrap/database.rs index b2f0d1c7..07bc5c7d 100644 --- a/synctv/src/bootstrap/database.rs +++ b/synctv/src/bootstrap/database.rs @@ -519,8 +519,10 @@ mod tests { #[test] fn effective_log_level_uses_synctv_config_for_database_policy() { let logging = LoggingOptions { + timezone: "UTC".to_string(), global: synctv_core::logging::ComponentLoggingOptions { name: "global".to_string(), + style: synctv_core::logging::LogStyle::Diagnostic, level: "debug".to_string(), targets: Vec::new(), format: "text".to_string(), diff --git a/synctv/src/config_env.rs b/synctv/src/config_env.rs index 689a4e9f..cc0e64ed 100644 --- a/synctv/src/config_env.rs +++ b/synctv/src/config_env.rs @@ -129,12 +129,7 @@ pub(crate) fn apply_env_overrides_with( } }; let apply_logging_env = - |service: &str, logging: &mut LoggingConfig| -> Result<(), ConfigError> { - let prefix = if service.is_empty() { - "SYNCTV_LOGGING".to_string() - } else { - format!("SYNCTV_{service}_LOGGING") - }; + |prefix: &str, logging: &mut LoggingConfig| -> Result<(), ConfigError> { if let Some(value) = get_env(&format!("{prefix}_LEVEL")) { logging.level = value; } @@ -712,14 +707,26 @@ pub(crate) fn apply_env_overrides_with( &mut config.webauthn.timeout_seconds, )?; - apply_logging_env("", &mut config.logging)?; - apply_logging_env("SERVER", &mut config.server.logging)?; - apply_logging_env("HEALTH", &mut config.health.logging)?; - apply_logging_env("METRICS", &mut config.metrics.logging)?; - apply_logging_env("CLUSTER", &mut config.cluster.logging)?; - apply_logging_env("MANAGEMENT", &mut config.management.logging)?; - apply_logging_env("LIVESTREAM", &mut config.livestream.logging)?; - apply_logging_env("WEBRTC", &mut config.webrtc.logging)?; + apply_logging_env("SYNCTV_LOGGING", &mut config.logging)?; + apply_logging_env("SYNCTV_SERVER_LOGGING", &mut config.server.logging)?; + apply_logging_env( + "SYNCTV_SERVER_ACCESS_LOG", + &mut config.server.access_log.logging, + )?; + env_override_bool( + "SYNCTV_SERVER_ACCESS_LOG_ENABLED", + &mut config.server.access_log.enabled, + )?; + env_override_parse( + "SYNCTV_SERVER_ACCESS_LOG_SLOW_REQUEST_THRESHOLD_MS", + &mut config.server.access_log.slow_request_threshold_ms, + )?; + apply_logging_env("SYNCTV_HEALTH_LOGGING", &mut config.health.logging)?; + apply_logging_env("SYNCTV_METRICS_LOGGING", &mut config.metrics.logging)?; + apply_logging_env("SYNCTV_CLUSTER_LOGGING", &mut config.cluster.logging)?; + apply_logging_env("SYNCTV_MANAGEMENT_LOGGING", &mut config.management.logging)?; + apply_logging_env("SYNCTV_LIVESTREAM_LOGGING", &mut config.livestream.logging)?; + apply_logging_env("SYNCTV_WEBRTC_LOGGING", &mut config.webrtc.logging)?; env_override_parse( "SYNCTV_LIVESTREAM_RTMP_PORT", @@ -1245,6 +1252,7 @@ pub(crate) fn resolve_owned_local_paths( for logging in [ &mut config.logging, &mut config.server.logging, + &mut config.server.access_log.logging, &mut config.health.logging, &mut config.metrics.logging, &mut config.cluster.logging, @@ -1474,6 +1482,13 @@ mod tests { ("SYNCTV_SERVER_LOGGING_LEVEL", "debug".to_string()), ("SYNCTV_SERVER_LOGGING_FORMAT", "json".to_string()), ("SYNCTV_SERVER_LOGGING_COLOR", "never".to_string()), + ("SYNCTV_SERVER_ACCESS_LOG_ENABLED", "false".to_string()), + ( + "SYNCTV_SERVER_ACCESS_LOG_SLOW_REQUEST_THRESHOLD_MS", + "2500".to_string(), + ), + ("SYNCTV_SERVER_ACCESS_LOG_LEVEL", "warn".to_string()), + ("SYNCTV_SERVER_ACCESS_LOG_FORMAT", "json".to_string()), ("SYNCTV_HEALTH_LOGGING_LEVEL", "error".to_string()), ("SYNCTV_HEALTH_LOGGING_FORMAT", "json".to_string()), ("SYNCTV_HEALTH_LOGGING_COLOR", "never".to_string()), @@ -1514,6 +1529,10 @@ mod tests { assert_eq!(config.server.logging.level, "debug"); assert_eq!(config.server.logging.format, "json"); assert!(matches!(config.server.logging.color, LogColor::Never)); + assert!(!config.server.access_log.enabled); + assert_eq!(config.server.access_log.slow_request_threshold_ms, 2500); + assert_eq!(config.server.access_log.logging.level, "warn"); + assert_eq!(config.server.access_log.logging.format, "json"); assert_eq!(config.health.logging.level, "error"); assert_eq!(config.health.logging.format, "json"); assert!(matches!(config.health.logging.color, LogColor::Never)); @@ -1546,6 +1565,10 @@ mod tests { path: "logs/server".to_string(), ..LogFileOutput::default() }); + config.server.access_log.logging.output = LogOutput::File(LogFileOutput { + path: "logs/access".to_string(), + ..LogFileOutput::default() + }); config.health.logging.output = LogOutput::File(LogFileOutput { path: "logs/health".to_string(), ..LogFileOutput::default() @@ -1575,6 +1598,13 @@ mod tests { output.path, data_dir.join("logs/server").display().to_string() ); + let LogOutput::File(output) = config.server.access_log.logging.output else { + panic!("access logging output should remain a file output"); + }; + assert_eq!( + output.path, + data_dir.join("logs/access").display().to_string() + ); let LogOutput::File(output) = config.health.logging.output else { panic!("health logging output should remain a file output"); }; diff --git a/synctv/src/config_loader.rs b/synctv/src/config_loader.rs index 71221574..95ee260b 100644 --- a/synctv/src/config_loader.rs +++ b/synctv/src/config_loader.rs @@ -921,6 +921,9 @@ mod tests { ); assert_eq!(config.logging.level, "info"); assert_eq!(config.server.logging.level, "info"); + assert!(config.server.access_log.enabled); + assert_eq!(config.server.access_log.slow_request_threshold_ms, 1000); + assert_eq!(config.server.access_log.logging.level, "info"); assert_eq!(config.health.logging.level, "info"); assert_eq!(config.metrics.logging.level, "warn"); assert_eq!(config.management.logging.level, "info"); diff --git a/synctv/src/resource_options.rs b/synctv/src/resource_options.rs index 85bf1917..a8f0bea7 100644 --- a/synctv/src/resource_options.rs +++ b/synctv/src/resource_options.rs @@ -8,10 +8,11 @@ use crate::bootstrap::{ SsrfOptions, }; use synctv_api::{ - ApiRuntimeSettings, ApiServerSettings, ClusterRuntimeSettings, ConnectionLimitSettings, - LivestreamRuntimeSettings, MetricsAuthMode, MetricsAuthSettings, MetricsKubernetesAuthSettings, - MetricsRuntimeSettings, ProxySliceCacheRuntimeSettings, RateLimitScopeRule, - RateLimitScopeStrategy, RedisRuntimeSettings, RequestRateLimitSettings, WebRtcRuntimeSettings, + AccessLogSettings, ApiRuntimeSettings, ApiServerSettings, ClusterRuntimeSettings, + ConnectionLimitSettings, LivestreamRuntimeSettings, MetricsAuthMode, MetricsAuthSettings, + MetricsKubernetesAuthSettings, MetricsRuntimeSettings, ProxySliceCacheRuntimeSettings, + RateLimitScopeRule, RateLimitScopeStrategy, RedisRuntimeSettings, RequestRateLimitSettings, + WebRtcRuntimeSettings, }; #[cfg(feature = "k8s")] use synctv_cluster::leader::K8sLeaderRuntimeOptions; @@ -20,7 +21,7 @@ use synctv_core::clock::{ ClockSyncOptions, ClockSyncProvider, ClockSyncSntpProviderOptions, TimeOptions, }; use synctv_core::logging::{ - ComponentLoggingOptions, LogColor, LogOutput, LogRotation, LoggingOptions, + ComponentLoggingOptions, LogColor, LogOutput, LogRotation, LogStyle, LoggingOptions, }; use synctv_core::service::{ LocalProviderHttpOptions, MediaProvidersOptions, PasskeyServiceOptions, @@ -40,8 +41,14 @@ use crate::app_config::{ pub fn logging_options(config: &AppConfig) -> LoggingOptions { LoggingOptions { + timezone: config.time.timezone.clone(), global: component_logging("global", &config.logging, Vec::new()), components: vec![ + access_component_logging( + "access", + &config.server.access_log.logging, + vec!["synctv::access".to_string()], + ), component_logging( "server", &config.server.logging, @@ -126,6 +133,7 @@ fn component_logging( }; ComponentLoggingOptions { name: name.to_string(), + style: LogStyle::Diagnostic, targets, level: config.level.clone(), format: config.format.clone(), @@ -138,6 +146,17 @@ fn component_logging( } } +fn access_component_logging( + name: &str, + config: &crate::app_config::LoggingConfig, + targets: Vec, +) -> ComponentLoggingOptions { + ComponentLoggingOptions { + style: LogStyle::Access, + ..component_logging(name, config, targets) + } +} + pub fn database_pool_options(config: &AppConfig) -> DatabasePoolOptions { DatabasePoolOptions { url: config.database.url.clone(), @@ -622,6 +641,10 @@ pub fn api_runtime_settings(config: &AppConfig) -> ApiRuntimeSettings { grpc_compression_enabled: config.server.grpc_compression_enabled, enable_reflection: config.server.enable_reflection, }, + access_log: AccessLogSettings { + enabled: config.server.access_log.enabled, + slow_request_threshold_ms: config.server.access_log.slow_request_threshold_ms, + }, request_rate_limits: request_rate_limit_settings(config), metrics: metrics_runtime_settings(config), cluster_enabled: config.cluster_runtime_enabled(), @@ -681,8 +704,11 @@ mod tests { #[test] fn global_logging_is_the_unscoped_output() { - let options = logging_options(&AppConfig::default()); + let mut config = AppConfig::default(); + config.time.timezone = "Asia/Shanghai".to_string(); + let options = logging_options(&config); + assert_eq!(options.timezone, "Asia/Shanghai"); assert_eq!(options.global.name, "global"); assert!(options.global.targets.is_empty()); assert!(options @@ -691,6 +717,18 @@ mod tests { .all(|component| !component.targets.is_empty())); } + #[test] + fn access_log_runtime_settings_follow_server_config() { + let mut config = AppConfig::default(); + config.server.access_log.enabled = false; + config.server.access_log.slow_request_threshold_ms = 2_500; + + let runtime = api_runtime_settings(&config); + + assert!(!runtime.access_log.enabled); + assert_eq!(runtime.access_log.slow_request_threshold_ms, 2_500); + } + #[test] fn cluster_logging_includes_application_bootstrap_target() { let options = logging_options(&AppConfig::default()); @@ -718,6 +756,8 @@ mod tests { .expect("logging component must exist") }; + assert!(targets_for("access").contains(&"synctv::access".to_string())); + assert!(!targets_for("server").contains(&"synctv::access".to_string())); assert!(targets_for("server").contains(&"synctv_api_http".to_string())); assert!(targets_for("metrics").contains(&"synctv_core::metrics".to_string())); assert!(targets_for("livestream").contains(&"synctv_livestream".to_string())); diff --git a/synctv/tests/cluster_startup_failure_tests.rs b/synctv/tests/cluster_startup_failure_tests.rs index 852f1660..802f829d 100644 --- a/synctv/tests/cluster_startup_failure_tests.rs +++ b/synctv/tests/cluster_startup_failure_tests.rs @@ -72,6 +72,7 @@ fn standalone_test_config() -> Config { advertise_host: String::new(), shutdown_drain_timeout_seconds: 30, logging: LoggingConfig::default(), + access_log: synctv::app_config::AccessLogConfig::default(), }, time: TimeConfig::default(), data_dir: default_data_dir().display().to_string(), @@ -127,6 +128,7 @@ fn cluster_test_config() -> Config { advertise_host: "127.0.0.1".to_string(), shutdown_drain_timeout_seconds: 30, logging: LoggingConfig::default(), + access_log: synctv::app_config::AccessLogConfig::default(), }, time: TimeConfig::default(), data_dir: default_data_dir().display().to_string(), diff --git a/synctv/tests/full_stack_e2e_tests.rs b/synctv/tests/full_stack_e2e_tests.rs index bfc203e6..6fc1cdec 100644 --- a/synctv/tests/full_stack_e2e_tests.rs +++ b/synctv/tests/full_stack_e2e_tests.rs @@ -550,8 +550,10 @@ static TEST_LOGGING: OnceLock = OnceLock::n fn ensure_test_logging() { TEST_LOGGING.get_or_init(|| { let logging = synctv_core::logging::LoggingOptions { + timezone: "UTC".to_string(), global: synctv_core::logging::ComponentLoggingOptions { name: "global".to_string(), + style: synctv_core::logging::LogStyle::Diagnostic, targets: Vec::new(), level: "debug".to_string(), format: "text".to_string(),