feat(logging): add structured transport access logs (#436)

## Summary

- add structured HTTP and gRPC access logs with request correlation,
status-aware levels, response bytes, and complete body lifecycle timing
- route diagnostic and access events through independently configurable
text or JSON outputs with non-blocking writers and dropped-line
accounting
- honor the configured IANA timezone across text, JSON, diagnostic, and
access logs with DST-aware RFC 3339 offsets
- keep HTTP metric labels bounded while logging concrete unmatched paths
without query strings
- reduce local development noise by defaulting global logs to info while
retaining component debug logs

## Behavior

- HTTP 2xx and 3xx complete at info, 4xx at warn, and 5xx or body
failures at error
- slow handlers are raised to warn; ordinary client response
cancellation remains debug
- gRPC completion waits for final trailers and records the canonical
gRPC status
- request IDs are validated, propagated in responses, and included in
mapped gRPC errors
- header latency and full response lifecycle latency are reported
separately
- log timestamps use the configured time.timezone and include a numeric
UTC offset

## Testing

- cargo test -p synctv-api-common transport_access_log --lib
- cargo test -p synctv-core logging::tests --lib
- cargo test -p synctv resource_options::tests --lib
- cargo clippy for affected crates and all targets with warnings denied
- cargo check --workspace --all-targets
- cargo fmt --all -- --check
- git diff --check
- local server smoke tests for request completion, request ID
propagation, byte counts, query omission, and Asia/Shanghai timestamps
in diagnostic and access logs
pull/437/head
zijiren 1 month ago committed by GitHub
parent 63c370876d
commit 882cbe09bb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194

@ -7952,6 +7952,7 @@ dependencies = [
"base64 0.23.1",
"bytes",
"chrono",
"chrono-tz",
"criterion",
"dashmap 7.0.0-rc2",
"failsafe",

@ -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}"; \

@ -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,

@ -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};

@ -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");
}
}

@ -6,6 +6,21 @@ pub struct AndroidAppAssociationSettings {
pub sha256_cert_fingerprints: Vec<String>,
}
#[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,

File diff suppressed because it is too large Load Diff

@ -293,23 +293,6 @@ const fn grpc_service_registration_plan(
}
}
fn request_targets_grpc_transport(
headers: &axum::http::HeaderMap,
) -> Result<bool, axum::http::header::ToStrError> {
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::sync::watch::Receiver<bool>>,
) -> 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(())

@ -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<synctv_api_common::impls::ApiError>) -> 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<dyn std::error::Error + Send + Sync>>;
@ -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(())
}
}

@ -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::<MatchedPath>().map_or_else(
|| "<unmatched>".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"));
}
}

@ -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<axum::http::HeaderName> =
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<axum::http::HeaderName> =
LazyLock::new(|| axum::http::HeaderName::from_static("x-frame-options"));

@ -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<AppState>,
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<AppState>, state: &AppState) -> anyhow::Result<axum::Router> {
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<AppState>, 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()),
)
}

@ -84,6 +84,7 @@ failsafe.workspace = true
# Time
chrono.workspace = true
chrono-tz.workspace = true
rsntp.workspace = true
# Logging

@ -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<Vec<ComponentErrorCounter>> = 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<String>,
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<ComponentLoggingOptions>,
}
@ -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<Utc>,
) -> std::fmt::Result {
write!(
writer,
"{}",
timestamp
.with_timezone(&timezone)
.format(LOG_TIMESTAMP_FORMAT)
)
}
#[cfg(test)]
fn format_log_timestamp(timezone: Tz, timestamp: DateTime<Utc>) -> 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<Vec<Box<dyn Layer<Registry> + 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<String> = 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(),

@ -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.

@ -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(),

@ -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(),
}
}
}

@ -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),

@ -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(),

@ -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");
};

@ -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");

@ -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<String>,
) -> 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()));

@ -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(),

@ -550,8 +550,10 @@ static TEST_LOGGING: OnceLock<synctv_core::logging::LoggingGuards> = 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(),

Loading…
Cancel
Save