From 1652b6076977ce0ff2eb7bba01fce1c2b5ba0048 Mon Sep 17 00:00:00 2001 From: zyycn Date: Sat, 11 Jul 2026 04:07:34 +0800 Subject: [PATCH] feat: harden logging and retention boundaries Co-authored-by: Codex --- README.md | 26 +- backend/src/api/middleware/trace.rs | 24 +- backend/src/api/router.rs | 9 +- backend/src/bootstrap/config.rs | 61 +- backend/src/bootstrap/services.rs | 35 +- backend/src/bootstrap/tasks/coordinator.rs | 2 + backend/src/bootstrap/tasks/retention_trim.rs | 41 +- backend/src/dispatch/recording.rs | 6 +- backend/src/fleet/pool/mod.rs | 28 +- backend/src/fleet/window.rs | 2 +- backend/src/infra/logging.rs | 549 +++++++++++++++--- backend/src/keys/service.rs | 2 +- backend/src/telemetry/account_usage/mod.rs | 31 + backend/src/telemetry/account_usage/store.rs | 59 +- backend/src/telemetry/recorder.rs | 6 +- .../src/upstream/openai/fingerprint/store.rs | 29 + .../tests/api/admin/settings_routes/routes.rs | 2 +- backend/tests/api/admin/usage_routes.rs | 2 +- backend/tests/api/middleware/trace.rs | 120 ++++ backend/tests/bootstrap/config/mod.rs | 44 +- backend/tests/dispatch/service.rs | 14 +- .../tests/dispatch/service/compact_routes.rs | 8 +- .../tests/dispatch/service/responses_http.rs | 2 +- .../dispatch/service/responses_recovery.rs | 4 +- .../dispatch/service/responses_websocket.rs | 2 +- .../tests/dispatch/service/usage_logging.rs | 20 +- backend/tests/infra/logging.rs | 108 +++- backend/tests/support/config.rs | 18 +- backend/tests/telemetry/account_usage/mod.rs | 1 + .../tests/telemetry/account_usage/store.rs | 49 ++ backend/tests/telemetry/mod.rs | 1 + .../tests/upstream/openai/fingerprint/mod.rs | 44 ++ deploy/README.md | 8 + deploy/config.example.yaml | 20 +- deploy/docker-compose.yml | 28 +- docs/database.md | 17 +- docs/deploy-ci-review.md | 4 + docs/logging-review.md | 40 +- 38 files changed, 1218 insertions(+), 248 deletions(-) create mode 100644 backend/tests/telemetry/account_usage/mod.rs create mode 100644 backend/tests/telemetry/account_usage/store.rs diff --git a/README.md b/README.md index 3b25776b7..d7a394426 100644 --- a/README.md +++ b/README.md @@ -47,7 +47,17 @@ redis: url: 'redis://127.0.0.1:6379/' logging: - directory: .runtime/logs + level: info + stdout: true + file: + enabled: true + directory: .runtime/logs + retention_days: 14 + max_file_size_mb: 20 + max_files: 20 + +telemetry: + enabled: true admin: default_password: '' @@ -87,10 +97,20 @@ database: url: 'postgres://codex_proxy:codex_proxy@postgres:5432/codex_proxy' redis: - url: 'redis://redis:6379/' + url: 'redis://:codex_proxy@redis:6379/' logging: - directory: /app/logs + level: info + stdout: true + file: + enabled: true + directory: /app/logs + retention_days: 14 + max_file_size_mb: 20 + max_files: 20 + +telemetry: + enabled: true ``` 构建并启动： diff --git a/backend/src/api/middleware/trace.rs b/backend/src/api/middleware/trace.rs index d8d8ded6e..e9b9cea5a 100644 --- a/backend/src/api/middleware/trace.rs +++ b/backend/src/api/middleware/trace.rs @@ -10,9 +10,9 @@ use tower_http::{ classify::{ServerErrorsAsFailures, ServerErrorsFailureClass, SharedClassifier}, trace::TraceLayer, }; -use tracing::{info, warn, Span}; +use tracing::{debug, info, warn, Span}; -use crate::api::middleware::request_id::RequestId; +use crate::api::middleware::request_id::{ClientIp, RequestId}; /// HTTP tracing layer type. pub type HttpTraceLayer = TraceLayer< @@ -36,27 +36,25 @@ pub fn http_trace_layer() -> HttpTraceLayer { fn make_http_span(request: &Request) -> Span { let request_id = request_id(request); + let client_ip = client_ip(request); let uri = sanitized_uri(request); tracing::info_span!( "http", request_id = request_id.as_deref(), + client_ip, method = %request.method(), uri = %uri ) } -fn on_http_request(request: &Request, _span: &Span) { - let request_id = request_id(request); - let uri = sanitized_uri(request); - info!( - request_id = request_id.as_deref(), - method = %request.method(), - uri = %uri, - "received HTTP request" - ); +fn on_http_request(_request: &Request, _span: &Span) { + debug!("received HTTP request"); } fn on_http_response(response: &Response, latency: Duration, _span: &Span) { + if response.status().is_server_error() { + return; + } info!( status = response.status().as_u16(), latency_ms = latency.as_millis(), @@ -83,6 +81,10 @@ fn request_id(request: &Request) -> Option { .map(|rid| rid.as_str().to_string()) } +fn client_ip(request: &Request) -> Option<&str> { + request.extensions().get::().map(ClientIp::as_str) +} + fn sanitized_uri(request: &Request) -> String { let uri = request.uri(); let path = uri.path(); diff --git a/backend/src/api/router.rs b/backend/src/api/router.rs index 537b6120d..b71f558a1 100644 --- a/backend/src/api/router.rs +++ b/backend/src/api/router.rs @@ -80,12 +80,15 @@ pub fn router() -> Router { /// 使用指定前端构建产物目录构造整个 HTTP 服务路由。 pub fn router_with_assets(dist_dir: impl AsRef) -> Router { - Router::new() - .route("/healthz", get(healthz)) + let traced_routes = Router::new() .merge(client::router::router()) .merge(admin::router::router()) .fallback_service(assets::spa_router(dist_dir)) - .layer(http_trace_layer()) + .layer(http_trace_layer()); + + Router::new() + .route("/healthz", get(healthz)) + .merge(traced_routes) .layer(middleware::from_fn(attach_request_id)) } diff --git a/backend/src/bootstrap/config.rs b/backend/src/bootstrap/config.rs index 5764a4f59..f3853d479 100644 --- a/backend/src/bootstrap/config.rs +++ b/backend/src/bootstrap/config.rs @@ -49,6 +49,9 @@ pub struct AppConfig { /// 日志配置。 #[serde(default)] pub logging: LoggingConfig, + /// PostgreSQL 遥测事实记录配置。 + #[serde(default)] + pub telemetry: TelemetryConfig, } /// HTTP 监听配置。 @@ -336,28 +339,70 @@ fn default_admin_password() -> String { String::new() } -/// 日志持久化配置。 +/// 应用结构化日志配置。 #[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] #[serde(deny_unknown_fields)] pub struct LoggingConfig { - /// 日志目录。 - pub directory: String, - /// 保留天数。 - pub retention_days: usize, - /// 是否启用日志。 - pub enabled: bool, + /// 默认日志级别；`RUST_LOG` 存在时优先使用环境变量。 + pub level: String, + /// 是否写入标准输出。 + pub stdout: bool, + /// 文件日志配置。 + pub file: FileLoggingConfig, } impl Default for LoggingConfig { fn default() -> Self { Self { + level: "info".to_string(), + stdout: true, + file: FileLoggingConfig::default(), + } + } +} + +/// 应用结构化文件日志配置。 +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct FileLoggingConfig { + /// 是否写入轮转文件。 + pub enabled: bool, + /// 日志目录。 + pub directory: String, + /// 按中国自然日计算的保留天数。 + pub retention_days: usize, + /// 单文件大小上限，单位 MiB。 + pub max_file_size_mb: u64, + /// 文件总数上限。 + pub max_files: usize, +} + +impl Default for FileLoggingConfig { + fn default() -> Self { + Self { + enabled: true, directory: ".runtime/logs".to_string(), retention_days: 14, - enabled: true, + max_file_size_mb: 20, + max_files: 20, } } } +/// PostgreSQL 遥测事实记录配置。 +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct TelemetryConfig { + /// 是否记录成功与失败代理事实。 + pub enabled: bool, +} + +impl Default for TelemetryConfig { + fn default() -> Self { + Self { enabled: true } + } +} + const CONFIG_FILE_ENV: &str = "CPR_CONFIG_FILE"; impl AppConfig { diff --git a/backend/src/bootstrap/services.rs b/backend/src/bootstrap/services.rs index 7a72fdb05..a1281f553 100644 --- a/backend/src/bootstrap/services.rs +++ b/backend/src/bootstrap/services.rs @@ -25,7 +25,7 @@ use crate::{ }, infra::{ database::connect, - logging::{init_tracing, LogGuard, RotationConfig}, + logging::{init_tracing, LogError, LogGuard, RotationConfig, TracingConfig}, paths::{ensure_data_dir, load_or_create_installation_id}, redis::RedisConnection, }, @@ -130,7 +130,7 @@ pub struct UsageRecordOptions { impl UsageRecordOptions { pub fn from_config(config: &AppConfig) -> Self { Self { - enabled: config.logging.enabled, + enabled: config.telemetry.enabled, capture_body: DEFAULT_USAGE_RECORD_CAPTURE_BODY, } } @@ -532,13 +532,32 @@ async fn build_router( )) } -fn init_logging(config: &AppConfig) -> Result, crate::infra::logging::LogError> { - if !config.logging.enabled { - return Ok(None); - } +fn init_logging(config: &AppConfig) -> Result { + const MEBIBYTE: u64 = 1024 * 1024; - let rotation = RotationConfig::new(&config.logging.directory, config.logging.retention_days); - Ok(Some(init_tracing(&rotation)?)) + let file = if config.logging.file.enabled { + let max_file_size_bytes = config + .logging + .file + .max_file_size_mb + .checked_mul(MEBIBYTE) + .ok_or(LogError::InvalidConfiguration( + "logging.file.max_file_size_mb is too large", + ))?; + Some(RotationConfig::new( + &config.logging.file.directory, + config.logging.file.retention_days, + max_file_size_bytes, + config.logging.file.max_files, + )) + } else { + None + }; + init_tracing(&TracingConfig::new( + &config.logging.level, + config.logging.stdout, + file, + )) } async fn wait_for_scheduled_restart() { diff --git a/backend/src/bootstrap/tasks/coordinator.rs b/backend/src/bootstrap/tasks/coordinator.rs index 12aafec06..a1b7a51f3 100644 --- a/backend/src/bootstrap/tasks/coordinator.rs +++ b/backend/src/bootstrap/tasks/coordinator.rs @@ -46,6 +46,8 @@ impl TaskCoordinator { stores.usage_records.clone(), stores.ops_errors.clone(), stores.request_buckets.clone(), + stores.account_usage.clone(), + stores.fingerprints.clone(), ) .start(), ); diff --git a/backend/src/bootstrap/tasks/retention_trim.rs b/backend/src/bootstrap/tasks/retention_trim.rs index 96b30605e..44064e096 100644 --- a/backend/src/bootstrap/tasks/retention_trim.rs +++ b/backend/src/bootstrap/tasks/retention_trim.rs @@ -1,11 +1,16 @@ //! PostgreSQL 增长表保留期清理任务。 +use std::num::NonZeroU32; + use chrono::Utc; use tracing::{info, warn}; -use crate::telemetry::{ - buckets::store::PgRequestBucketStore, ops::store::PgOpsErrorLogStore, - usage::store::PgUsageRecordStore, +use crate::{ + telemetry::{ + account_usage::store::PgAccountUsageStore, buckets::store::PgRequestBucketStore, + ops::store::PgOpsErrorLogStore, usage::store::PgUsageRecordStore, + }, + upstream::openai::fingerprint::PgFingerprintStore, }; use super::{ @@ -14,11 +19,15 @@ use super::{ }; const RETENTION_TRIM_INTERVAL_SECS: u64 = 60 * 60; +const ACCOUNT_MODEL_USAGE_LIMIT: NonZeroU32 = NonZeroU32::new(100).unwrap(); +const FINGERPRINT_HISTORY_LIMIT: NonZeroU32 = NonZeroU32::new(100).unwrap(); pub struct RetentionTrimTask { usage_records: PgUsageRecordStore, ops_errors: PgOpsErrorLogStore, buckets: PgRequestBucketStore, + account_usage: PgAccountUsageStore, + fingerprints: PgFingerprintStore, } impl RetentionTrimTask { @@ -26,11 +35,15 @@ impl RetentionTrimTask { usage_records: PgUsageRecordStore, ops_errors: PgOpsErrorLogStore, buckets: PgRequestBucketStore, + account_usage: PgAccountUsageStore, + fingerprints: PgFingerprintStore, ) -> Self { Self { usage_records, ops_errors, buckets, + account_usage, + fingerprints, } } @@ -65,6 +78,28 @@ impl PeriodicTaskRunner for RetentionTrimTask { Ok(_) => {} Err(error) => warn!(error = %error, "清理请求时间桶失败"), } + match self + .account_usage + .trim_model_usage_to_limit(ACCOUNT_MODEL_USAGE_LIMIT) + .await + { + Ok(deleted) if deleted > 0 => { + info!(deleted, "已清理超限账号模型用量统计") + } + Ok(_) => {} + Err(error) => warn!(error = %error, "清理超限账号模型用量统计失败"), + } + match self + .fingerprints + .trim_update_history_to_limit(FINGERPRINT_HISTORY_LIMIT) + .await + { + Ok(deleted) if deleted > 0 => { + info!(deleted, "已清理超限指纹更新历史") + } + Ok(_) => {} + Err(error) => warn!(error = %error, "清理超限指纹更新历史失败"), + } }) } } diff --git a/backend/src/dispatch/recording.rs b/backend/src/dispatch/recording.rs index 2854c7682..0c9621840 100644 --- a/backend/src/dispatch/recording.rs +++ b/backend/src/dispatch/recording.rs @@ -179,7 +179,7 @@ pub(super) async fn record_response_upstream_error_event( ); event.metadata = metadata; if let Err(error) = record.recorder.record_error(event).await { - tracing::warn!(account_id = %record.account_id, error = %error, "failed to record upstream error event"); + tracing::error!(account_id = %record.account_id, error = %error, "failed to record upstream error event"); } } @@ -246,7 +246,7 @@ pub(super) async fn record_prefetched_response_stream_failure_event( } event.metadata = metadata; if let Err(error) = record.recorder.record_error(event).await { - tracing::warn!(account_id = %record.account_id, error = %error, "failed to record prefetched stream error event"); + tracing::error!(account_id = %record.account_id, error = %error, "failed to record prefetched stream error event"); } } @@ -538,7 +538,7 @@ pub(super) async fn record_live_response_stream_event( event.latency_ms = Some(elapsed_millis_i64(context.started_at)); event.metadata = metadata; if let Err(error) = context.recorder.record_error(event).await { - tracing::warn!(account_id = %context.account_id, error = %error, "failed to record live response stream error event"); + tracing::error!(account_id = %context.account_id, error = %error, "failed to record live response stream error event"); } } } diff --git a/backend/src/fleet/pool/mod.rs b/backend/src/fleet/pool/mod.rs index 0bb6858db..3f9b41ccd 100644 --- a/backend/src/fleet/pool/mod.rs +++ b/backend/src/fleet/pool/mod.rs @@ -202,7 +202,7 @@ impl AccountPoolService { return; } if let Err(error) = self.usage_store.record_request(account_id).await { - tracing::warn!( + tracing::error!( account_id, error = %error, "failed to persist account request usage" @@ -227,7 +227,7 @@ impl AccountPoolService { ) .await { - tracing::warn!( + tracing::error!( account_id, model, error = %error, @@ -241,7 +241,7 @@ impl AccountPoolService { let persisted = match self.store.set_status(account_id, status).await { Ok(persisted) => persisted, Err(error) => { - tracing::warn!( + tracing::error!( account_id, error = %error, "failed to persist account status" @@ -258,7 +258,7 @@ impl AccountPoolService { let persisted = match self.store.mark_quota_limited_until(account_id, until).await { Ok(persisted) => persisted, Err(error) => { - tracing::warn!( + tracing::error!( account_id, error = %error, "failed to persist quota cooldown state" @@ -301,7 +301,7 @@ impl AccountPoolService { .sync_runtime_account_state(&refreshed.account) .await { - tracing::warn!( + tracing::error!( account_id = %refreshed.account.id, error = %error, "failed to persist refreshed runtime account state" @@ -316,7 +316,7 @@ impl AccountPoolService { ) .await { - tracing::warn!( + tracing::error!( account_id = %refreshed.account.id, error = %error, "failed to persist refreshed runtime account usage window" @@ -380,7 +380,7 @@ impl AccountPoolService { .record_usage_delta(account_id, persisted_usage) .await { - tracing::warn!( + tracing::error!( account_id, error = %error, "failed to persist account token usage" @@ -400,7 +400,7 @@ impl AccountPoolService { ) .await { - tracing::warn!( + tracing::error!( account_id, model, error = %error, @@ -434,7 +434,7 @@ impl AccountPoolService { ..AccountUsageDelta::default() }; if let Err(error) = self.usage_store.record_usage_delta(account_id, usage).await { - tracing::warn!( + tracing::error!( account_id, error = %error, "failed to persist empty response usage" @@ -452,7 +452,7 @@ impl AccountPoolService { ) .await { - tracing::warn!( + tracing::error!( account_id, model, error = %error, @@ -498,7 +498,7 @@ impl AccountPoolService { Ok(Some(quota_json)) => serde_json::from_str::("a_json).ok(), Ok(None) => None, Err(error) => { - tracing::warn!( + tracing::error!( account_id = %account_id, error = %error, "failed to read existing quota json before passive rate-limit sync" @@ -530,7 +530,7 @@ impl AccountPoolService { { Ok(persisted) => persisted, Err(error) => { - tracing::warn!( + tracing::error!( account_id, error = %error, "failed to persist verified quota snapshot" @@ -545,7 +545,7 @@ impl AccountPoolService { .sync_rate_limit_window(account_id, reset_at, limit_window_seconds) .await { - tracing::warn!( + tracing::error!( account_id, error = %error, "failed to persist verified quota window" @@ -609,7 +609,7 @@ impl AccountPoolService { { Ok(persisted) => persisted, Err(error) => { - tracing::warn!( + tracing::error!( account_id, error = %error, "failed to persist Cloudflare cooldown state" diff --git a/backend/src/fleet/window.rs b/backend/src/fleet/window.rs index 3591f8241..e64765413 100644 --- a/backend/src/fleet/window.rs +++ b/backend/src/fleet/window.rs @@ -1,3 +1,3 @@ //! 账号配额窗口规则。 -pub(crate) use crate::telemetry::account_usage::store::should_reset_usage_window; +pub(crate) use crate::telemetry::account_usage::should_reset_usage_window; diff --git a/backend/src/infra/logging.rs b/backend/src/infra/logging.rs index b4fa39020..d60935e31 100644 --- a/backend/src/infra/logging.rs +++ b/backend/src/infra/logging.rs @@ -1,16 +1,20 @@ -//! 日志初始化与文件轮转。 +//! 结构化日志初始化、输出与文件轮转。 use std::{ - fmt, fs, + backtrace::Backtrace, + env, fmt, fs, fs::{File, OpenOptions}, - io, - io::Write, + io::{self, Write}, + panic, path::{Path, PathBuf}, + sync::mpsc, + thread, + time::Duration as StdDuration, }; -use chrono::{DateTime, Utc}; +use chrono::{DateTime, Duration, NaiveDate, Utc}; use thiserror::Error; -use tracing_appender::non_blocking::WorkerGuard; +use tracing_appender::non_blocking::{ErrorCounter, NonBlockingBuilder, WorkerGuard}; use tracing_subscriber::{ fmt::{self as tracing_fmt, format::Writer, time::FormatTime}, layer::SubscriberExt, @@ -22,24 +26,89 @@ use crate::infra::time::{china_date, china_rfc3339_millis}; const LOG_PREFIX: &str = "codex-proxy-rs"; const LOG_SUFFIX: &str = "log"; +const DROP_MONITOR_INTERVAL: StdDuration = StdDuration::from_secs(10); -/// 日志轮转配置。 +/// 文件日志轮转配置。 #[derive(Debug, Clone)] pub struct RotationConfig { /// 日志目录。 pub directory: PathBuf, - /// 保留周期。 + /// 按中国自然日计算的保留天数。 pub retention_days: usize, + /// 单文件大小上限。 + pub max_file_size_bytes: u64, + /// 文件总数上限。 + pub max_files: usize, } impl RotationConfig { /// 创建轮转配置。 - pub fn new(directory: impl AsRef, retention_days: usize) -> Self { + pub fn new( + directory: impl AsRef, + retention_days: usize, + max_file_size_bytes: u64, + max_files: usize, + ) -> Self { Self { directory: directory.as_ref().to_path_buf(), retention_days, + max_file_size_bytes, + max_files, } } + + fn validate(&self) -> Result<(), LogError> { + if self.retention_days == 0 { + return Err(LogError::InvalidConfiguration( + "logging.file.retention_days must be greater than zero", + )); + } + if self.max_file_size_bytes == 0 { + return Err(LogError::InvalidConfiguration( + "logging.file.max_file_size_mb must be greater than zero", + )); + } + if self.max_files == 0 { + return Err(LogError::InvalidConfiguration( + "logging.file.max_files must be greater than zero", + )); + } + Ok(()) + } +} + +/// tracing 输出配置。 +#[derive(Debug, Clone)] +pub struct TracingConfig { + /// 默认过滤级别或指令。 + pub level: String, + /// 是否写入标准输出。 + pub stdout: bool, + /// 可选文件输出。 + pub file: Option, +} + +impl TracingConfig { + /// 创建 tracing 输出配置。 + pub fn new(level: impl Into, stdout: bool, file: Option) -> Self { + Self { + level: level.into(), + stdout, + file, + } + } + + fn validate(&self) -> Result<(), LogError> { + if !self.stdout && self.file.is_none() { + return Err(LogError::InvalidConfiguration( + "at least one of logging.stdout or logging.file.enabled must be true", + )); + } + if let Some(file) = &self.file { + file.validate()?; + } + Ok(()) + } } /// 日志初始化错误。 @@ -48,94 +117,198 @@ pub enum LogError { /// 日志 IO 错误。 #[error("log io error: {0}")] Io(#[from] io::Error), + /// 日志配置非法。 + #[error("invalid logging configuration: {0}")] + InvalidConfiguration(&'static str), + /// 日志过滤指令非法。 + #[error("invalid logging filter `{filter}`: {message}")] + InvalidFilter { + /// 原始过滤指令。 + filter: String, + /// 解析错误。 + message: String, + }, /// tracing subscriber 已初始化。 #[error("global tracing subscriber is already initialized")] SubscriberAlreadyInitialized, } -/// 日志守卫。 +/// 日志守卫；drop 时先停止丢弃监测，再刷出两个 non-blocking writer。 pub struct LogGuard { - _guard: WorkerGuard, + _drop_monitor: DroppedLogMonitor, + _writer_guards: Vec, } -/// 按中国自然日写入的日志 appender。 +/// 按中国自然日和大小写入的日志 appender。 pub struct ChinaDailyFileAppender { - directory: PathBuf, - retention_days: usize, + config: RotationConfig, current_date: String, + current_segment: usize, + bytes_written: u64, writer: File, } /// 构造文件 appender。 pub fn build_file_appender(config: &RotationConfig) -> Result { + config.validate()?; fs::create_dir_all(&config.directory)?; - ChinaDailyFileAppender::open(&config.directory, config.retention_days, Utc::now()) + ChinaDailyFileAppender::open(config.clone(), Utc::now()) } /// 初始化 tracing。 -pub fn init_tracing(config: &RotationConfig) -> Result { - let appender = build_file_appender(config)?; - let (writer, guard) = tracing_appender::non_blocking(appender); - let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); +pub fn init_tracing(config: &TracingConfig) -> Result { + config.validate()?; - tracing_subscriber::registry() + let filter_directive = env::var("RUST_LOG").unwrap_or_else(|_| config.level.clone()); + let filter = + EnvFilter::try_new(&filter_directive).map_err(|error| LogError::InvalidFilter { + filter: filter_directive, + message: error.to_string(), + })?; + + let mut writer_guards = Vec::new(); + let mut counters = Vec::new(); + + let stdout_writer = config.stdout.then(|| { + let (writer, guard) = NonBlockingBuilder::default() + .lossy(true) + .thread_name("tracing-stdout") + .finish(io::stdout()); + counters.push(NamedErrorCounter::new("stdout", writer.error_counter())); + writer_guards.push(guard); + writer + }); + + let file_writer = config + .file + .as_ref() + .map(build_file_appender) + .transpose()? + .map(|appender| { + let (writer, guard) = NonBlockingBuilder::default() + .lossy(true) + .thread_name("tracing-file") + .finish(appender); + counters.push(NamedErrorCounter::new("file", writer.error_counter())); + writer_guards.push(guard); + writer + }); + + let stdout_layer = stdout_writer.map(json_layer); + let file_layer = file_writer.map(json_layer); + let drop_monitor = DroppedLogMonitor::start(counters)?; + + let init_result = tracing_subscriber::registry() .with(filter) - .with( - tracing_fmt::layer() - .json() - .with_timer(ChinaLogTimer) - .with_writer(writer) - .with_target(true) - .with_level(true) - .with_file(true) - .with_line_number(true) - .with_thread_ids(true) - .with_thread_names(true) - .with_current_span(true) - .with_span_list(true), - ) + .with(stdout_layer) + .with(file_layer) .try_init() - .map_err(|_| LogError::SubscriberAlreadyInitialized)?; + .map_err(|_| LogError::SubscriberAlreadyInitialized); + if let Err(error) = init_result { + drop(drop_monitor); + drop(writer_guards); + return Err(error); + } + install_panic_hook(); - Ok(LogGuard { _guard: guard }) + Ok(LogGuard { + _drop_monitor: drop_monitor, + _writer_guards: writer_guards, + }) +} + +fn install_panic_hook() { + let previous_hook = panic::take_hook(); + panic::set_hook(Box::new(move |panic_info| { + tracing::error!( + panic = %panic_info, + backtrace = %Backtrace::force_capture(), + "process panicked" + ); + previous_hook(panic_info); + })); +} + +fn json_layer( + writer: tracing_appender::non_blocking::NonBlocking, +) -> tracing_subscriber::fmt::Layer< + S, + tracing_subscriber::fmt::format::JsonFields, + tracing_subscriber::fmt::format::Format, + tracing_appender::non_blocking::NonBlocking, +> +where + S: tracing::Subscriber + for<'span> tracing_subscriber::registry::LookupSpan<'span>, +{ + tracing_fmt::layer() + .json() + .with_timer(ChinaLogTimer) + .with_writer(writer) + .with_target(true) + .with_level(true) + .with_file(true) + .with_line_number(true) + .with_thread_ids(true) + .with_thread_names(true) + .with_current_span(true) + .with_span_list(true) } impl ChinaDailyFileAppender { - fn open( - directory: impl AsRef, - retention_days: usize, - now: DateTime, - ) -> Result { - let directory = directory.as_ref().to_path_buf(); + fn open(config: RotationConfig, now: DateTime) -> Result { + cleanup_managed_logs(&config, now)?; let current_date = china_date(&now); - let writer = open_log_file(&directory, ¤t_date)?; - cleanup_old_logs(&directory, retention_days)?; + let (current_segment, bytes_written) = writable_segment(&config, ¤t_date)?; + let writer = open_log_file(&config.directory, ¤t_date, current_segment)?; + cleanup_managed_logs(&config, now)?; Ok(Self { - directory, - retention_days, + config, current_date, + current_segment, + bytes_written, writer, }) } - fn rollover_if_needed(&mut self) -> io::Result<()> { - let current_date = china_date(&Utc::now()); - if current_date == self.current_date { + fn rollover_if_needed(&mut self, incoming_bytes: usize) -> io::Result<()> { + let now = Utc::now(); + let current_date = china_date(&now); + let date_changed = current_date != self.current_date; + let size_exceeded = !date_changed + && self.bytes_written > 0 + && self + .bytes_written + .saturating_add(u64::try_from(incoming_bytes).unwrap_or(u64::MAX)) + > self.config.max_file_size_bytes; + if !date_changed && !size_exceeded { return Ok(()); } self.writer.flush()?; - self.writer = open_log_file(&self.directory, ¤t_date)?; - self.current_date = current_date; - cleanup_old_logs(&self.directory, self.retention_days)?; - Ok(()) + if date_changed { + self.current_date = current_date; + self.current_segment = 0; + } else { + self.current_segment = self.current_segment.saturating_add(1); + } + self.writer = open_log_file( + &self.config.directory, + &self.current_date, + self.current_segment, + )?; + self.bytes_written = self.writer.metadata()?.len(); + cleanup_managed_logs(&self.config, now).map_err(log_error_into_io) } } impl Write for ChinaDailyFileAppender { fn write(&mut self, buf: &[u8]) -> io::Result { - self.rollover_if_needed()?; - self.writer.write(buf) + self.rollover_if_needed(buf.len())?; + let written = self.writer.write(buf)?; + self.bytes_written = self + .bytes_written + .saturating_add(u64::try_from(written).unwrap_or(u64::MAX)); + Ok(written) } fn flush(&mut self) -> io::Result<()> { @@ -156,38 +329,268 @@ fn china_log_timestamp(now: DateTime) -> String { china_rfc3339_millis(&now) } -fn open_log_file(directory: &Path, date: &str) -> io::Result { +fn writable_segment(config: &RotationConfig, date: &str) -> io::Result<(usize, u64)> { + let latest = managed_log_files(&config.directory)? + .into_iter() + .filter(|file| file.date.to_string() == date) + .max_by_key(|file| file.segment); + let Some(latest) = latest else { + return Ok((0, 0)); + }; + let size = latest.path.metadata()?.len(); + if size >= config.max_file_size_bytes { + Ok((latest.segment.saturating_add(1), 0)) + } else { + Ok((latest.segment, size)) + } +} + +fn open_log_file(directory: &Path, date: &str, segment: usize) -> io::Result { OpenOptions::new() .create(true) .append(true) - .open(directory.join(log_file_name(date))) + .open(directory.join(log_file_name(date, segment))) } -fn log_file_name(date: &str) -> String { - format!("{LOG_PREFIX}.{date}.{LOG_SUFFIX}") -} - -fn cleanup_old_logs(directory: &Path, retention_days: usize) -> io::Result<()> { - if retention_days == 0 { - return Ok(()); +fn log_file_name(date: &str, segment: usize) -> String { + if segment == 0 { + format!("{LOG_PREFIX}.{date}.{LOG_SUFFIX}") + } else { + format!("{LOG_PREFIX}.{date}.{segment}.{LOG_SUFFIX}") } +} - let mut logs = fs::read_dir(directory)? +#[derive(Debug)] +struct ManagedLogFile { + date: NaiveDate, + segment: usize, + path: PathBuf, +} + +fn managed_log_files(directory: &Path) -> io::Result> { + Ok(fs::read_dir(directory)? .filter_map(Result::ok) .filter_map(|entry| { let file_name = entry.file_name().to_string_lossy().into_owned(); - is_managed_log_file(&file_name).then_some((file_name, entry.path())) + parse_managed_log_name(&file_name).map(|(date, segment)| ManagedLogFile { + date, + segment, + path: entry.path(), + }) }) - .collect::>(); - logs.sort_by(|left, right| right.0.cmp(&left.0)); + .collect()) +} - for (_, path) in logs.into_iter().skip(retention_days) { - fs::remove_file(path)?; +fn parse_managed_log_name(file_name: &str) -> Option<(NaiveDate, usize)> { + let body = file_name + .strip_prefix(&format!("{LOG_PREFIX}."))? + .strip_suffix(&format!(".{LOG_SUFFIX}"))?; + let (date, segment) = match body.split_once('.') { + Some((date, segment)) => (date, segment.parse().ok()?), + None => (body, 0), + }; + NaiveDate::parse_from_str(date, "%Y-%m-%d") + .ok() + .map(|date| (date, segment)) +} + +fn cleanup_managed_logs(config: &RotationConfig, now: DateTime) -> Result<(), LogError> { + let today = NaiveDate::parse_from_str(&china_date(&now), "%Y-%m-%d") + .map_err(|_| LogError::InvalidConfiguration("failed to derive current China date"))?; + let retention_offset = i64::try_from(config.retention_days.saturating_sub(1)) + .map_err(|_| LogError::InvalidConfiguration("logging.file.retention_days is too large"))?; + let cutoff = today + .checked_sub_signed(Duration::days(retention_offset)) + .ok_or(LogError::InvalidConfiguration( + "logging.file.retention_days is too large", + ))?; + + let mut retained = Vec::new(); + for file in managed_log_files(&config.directory)? { + if file.date < cutoff { + fs::remove_file(file.path)?; + } else { + retained.push(file); + } + } + retained.sort_by(|left, right| { + right + .date + .cmp(&left.date) + .then_with(|| right.segment.cmp(&left.segment)) + }); + for file in retained.into_iter().skip(config.max_files) { + fs::remove_file(file.path)?; } Ok(()) } -fn is_managed_log_file(file_name: &str) -> bool { - file_name.starts_with(&format!("{LOG_PREFIX}.")) - && file_name.ends_with(&format!(".{LOG_SUFFIX}")) +fn log_error_into_io(error: LogError) -> io::Error { + match error { + LogError::Io(error) => error, + other => io::Error::other(other), + } +} + +#[derive(Debug)] +struct NamedErrorCounter { + output: &'static str, + counter: ErrorCounter, + last_reported: usize, +} + +impl NamedErrorCounter { + fn new(output: &'static str, counter: ErrorCounter) -> Self { + Self { + output, + counter, + last_reported: 0, + } + } +} + +struct DroppedLogMonitor { + shutdown_tx: mpsc::Sender<()>, + handle: Option>, +} + +impl DroppedLogMonitor { + fn start(mut counters: Vec) -> io::Result { + let (shutdown_tx, shutdown_rx) = mpsc::channel(); + let handle = thread::Builder::new() + .name("tracing-drop-monitor".to_string()) + .spawn(move || loop { + match shutdown_rx.recv_timeout(DROP_MONITOR_INTERVAL) { + Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => { + report_dropped_logs(&mut counters, &mut io::stderr().lock()); + break; + } + Err(mpsc::RecvTimeoutError::Timeout) => { + report_dropped_logs(&mut counters, &mut io::stderr().lock()); + } + } + })?; + Ok(Self { + shutdown_tx, + handle: Some(handle), + }) + } +} + +impl Drop for DroppedLogMonitor { + fn drop(&mut self) { + let _ = self.shutdown_tx.send(()); + if let Some(handle) = self.handle.take() { + let _ = handle.join(); + } + } +} + +fn report_dropped_logs(counters: &mut [NamedErrorCounter], writer: &mut impl Write) { + for counter in counters { + let total = counter.counter.dropped_lines(); + if total <= counter.last_reported { + continue; + } + let dropped = total - counter.last_reported; + counter.last_reported = total; + let warning = serde_json::json!({ + "timestamp": china_log_timestamp(Utc::now()), + "level": "ERROR", + "target": "codex_proxy_rs::infra::logging", + "fields": { + "message": "non-blocking logging output dropped events", + "output": counter.output, + "dropped": dropped, + "dropped_total": total, + } + }); + let _ = writeln!(writer, "{warning}"); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + struct FirstWriteBlocking { + started_tx: mpsc::SyncSender<()>, + release_rx: Option>, + } + + impl Write for FirstWriteBlocking { + fn write(&mut self, buffer: &[u8]) -> io::Result { + if let Some(release_rx) = self.release_rx.take() { + self.started_tx + .send(()) + .map_err(|error| io::Error::other(error.to_string()))?; + release_rx + .recv() + .map_err(|error| io::Error::other(error.to_string()))?; + } + Ok(buffer.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + } + + #[test] + fn tracing_config_should_reject_disabled_outputs() { + let config = TracingConfig::new("info", false, None); + + assert!(config.validate().is_err()); + } + + #[test] + fn tracing_config_should_accept_each_output_independently() { + let directory = tempfile::tempdir().unwrap(); + let stdout_only = TracingConfig::new("info", true, None); + let file_only = TracingConfig::new( + "info", + false, + Some(RotationConfig::new(directory.path(), 14, 1024, 20)), + ); + + assert_eq!( + (stdout_only.validate().is_ok(), file_only.validate().is_ok()), + (true, true) + ); + } + + #[test] + fn dropped_log_report_should_emit_emergency_json_for_full_buffer() { + let (started_tx, started_rx) = mpsc::sync_channel(0); + let (release_tx, release_rx) = mpsc::sync_channel(0); + let (mut writer, guard) = NonBlockingBuilder::default() + .lossy(true) + .buffered_lines_limit(1) + .finish(FirstWriteBlocking { + started_tx, + release_rx: Some(release_rx), + }); + let counter = writer.error_counter(); + + writer.write_all(b"first\n").unwrap(); + started_rx.recv().unwrap(); + writer.write_all(b"second\n").unwrap(); + writer.write_all(b"dropped\n").unwrap(); + + let mut output = Vec::new(); + report_dropped_logs(&mut [NamedErrorCounter::new("test", counter)], &mut output); + release_tx.send(()).unwrap(); + drop(writer); + drop(guard); + + let warning: serde_json::Value = serde_json::from_slice(&output).unwrap(); + assert_eq!( + ( + warning["level"].as_str(), + warning["fields"]["output"].as_str(), + warning["fields"]["dropped"].as_u64(), + ), + (Some("ERROR"), Some("test"), Some(1)) + ); + } } diff --git a/backend/src/keys/service.rs b/backend/src/keys/service.rs index 8b56f9a6f..2cc98e4d3 100644 --- a/backend/src/keys/service.rs +++ b/backend/src/keys/service.rs @@ -86,7 +86,7 @@ impl KeyVerifier { pub async fn flush_pending_last_used(&self) { let updates = self.take_pending_last_used(); if let Err(error) = self.store.touch_last_used_batch(&updates).await { - tracing::warn!(error = %error, "failed to flush client key last_used_at batch"); + tracing::error!(error = %error, "failed to flush client key last_used_at batch"); self.pending_last_used .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) diff --git a/backend/src/telemetry/account_usage/mod.rs b/backend/src/telemetry/account_usage/mod.rs index f839156b7..74329dec3 100644 --- a/backend/src/telemetry/account_usage/mod.rs +++ b/backend/src/telemetry/account_usage/mod.rs @@ -1,4 +1,35 @@ //! 账号累计用量与时间桶查询。 +use chrono::{DateTime, Utc}; + pub mod query; pub mod store; + +/// 根据额度窗口边界漂移判断是否重置本地窗口计数。 +pub fn should_reset_usage_window( + existing_reset_at: Option>, + existing_limit_window_seconds: Option, + new_reset_at: DateTime, + new_limit_window_seconds: Option, +) -> bool { + let Some(existing_reset_at) = existing_reset_at else { + return false; + }; + if existing_reset_at == new_reset_at { + return false; + } + + let drift = existing_reset_at + .signed_duration_since(new_reset_at) + .num_seconds() + .unsigned_abs(); + let window_seconds = new_limit_window_seconds + .or(existing_limit_window_seconds) + .unwrap_or(0); + let threshold = if window_seconds > 0 { + window_seconds / 2 + } else { + 3_600 + }; + drift >= threshold +} diff --git a/backend/src/telemetry/account_usage/store.rs b/backend/src/telemetry/account_usage/store.rs index 7f891356c..635274a04 100644 --- a/backend/src/telemetry/account_usage/store.rs +++ b/backend/src/telemetry/account_usage/store.rs @@ -1,6 +1,6 @@ //! PostgreSQL 账号用量查询存储。 -use std::collections::HashMap; +use std::{collections::HashMap, num::NonZeroU32}; use async_trait::async_trait; use chrono::{DateTime, Utc}; @@ -13,6 +13,8 @@ use crate::infra::{ time::parse_rfc3339_utc as parse_rfc3339, }; +use super::should_reset_usage_window; + const RECORD_USAGE_SQL: &str = r" insert into account_usage ( account_id, request_count, empty_response_count, @@ -387,6 +389,33 @@ impl PgAccountUsageStore { &self.pool } + /// 将每个账号的模型用量统计裁剪为最近使用的指定行数。 + pub async fn trim_model_usage_to_limit( + &self, + max_models_per_account: NonZeroU32, + ) -> PgAccountUsageStoreResult { + let result = sqlx::query( + r" + delete from account_model_usage as target + using ( + select account_id, model + from ( + select account_id, model, + row_number() over (partition by account_id order by last_used_at desc nulls last, model) as position + from account_model_usage + ) as ranked + where position > $1 + ) as expired + where target.account_id = expired.account_id + and target.model = expired.model + ", + ) + .bind(i64::from(max_models_per_account.get())) + .execute(&self.pool) + .await?; + Ok(result.rows_affected()) + } + async fn load_snapshots( &self, account_ids: &[String], @@ -732,34 +761,6 @@ fn usage_snapshot_from_row(row: &PgRow) -> AccountUsageSnapshot { } } -pub fn should_reset_usage_window( - existing_reset_at: Option>, - existing_limit_window_seconds: Option, - new_reset_at: DateTime, - new_limit_window_seconds: Option, -) -> bool { - let Some(existing_reset_at) = existing_reset_at else { - return false; - }; - if existing_reset_at == new_reset_at { - return false; - } - - let drift = existing_reset_at - .signed_duration_since(new_reset_at) - .num_seconds() - .unsigned_abs(); - let window_seconds = new_limit_window_seconds - .or(existing_limit_window_seconds) - .unwrap_or(0); - let threshold = if window_seconds > 0 { - window_seconds / 2 - } else { - 3_600 - }; - drift >= threshold -} - fn u64_to_i64_saturating(value: u64) -> i64 { value.min(i64::MAX as u64) as i64 } diff --git a/backend/src/telemetry/recorder.rs b/backend/src/telemetry/recorder.rs index ac53af0d2..d9a9b3acf 100644 --- a/backend/src/telemetry/recorder.rs +++ b/backend/src/telemetry/recorder.rs @@ -61,7 +61,7 @@ impl Recorder { return Ok(()); } if !is_usage_fact(&event) { - tracing::warn!( + tracing::error!( usage_record_id = %event.id, request_id = event.request_id.as_deref().unwrap_or(""), account_id = event.account_id, @@ -115,7 +115,7 @@ pub(crate) async fn record_dispatch_error_event(record: DispatchErrorLogRecord<' event.metadata = metadata; if let Err(error) = record.recorder.record_error(event).await { - tracing::warn!( + tracing::error!( account_id = record.account_id.unwrap_or(""), error = %error, "failed to record dispatch error event" @@ -163,7 +163,7 @@ pub(crate) async fn record_response_event(record: ResponseUsageRecord<'_>) { event.metadata = metadata; if let Err(error) = record.recorder.record_usage(event).await { - tracing::warn!(account_id = %record.account_id, error = %error, "failed to record response event"); + tracing::error!(account_id = %record.account_id, error = %error, "failed to record response event"); } } diff --git a/backend/src/upstream/openai/fingerprint/store.rs b/backend/src/upstream/openai/fingerprint/store.rs index cb3dc7bf4..49cc08c2b 100644 --- a/backend/src/upstream/openai/fingerprint/store.rs +++ b/backend/src/upstream/openai/fingerprint/store.rs @@ -1,5 +1,7 @@ //! OpenAI 客户端指纹 PostgreSQL 存储。 +use std::num::NonZeroU32; + use chrono::{DateTime, Utc}; use indexmap::IndexMap; use serde::{Deserialize, Serialize}; @@ -170,6 +172,33 @@ impl PgFingerprintStore { .await?; Ok(()) } + + /// 将指纹更新历史裁剪为最新的指定条数。 + pub async fn trim_update_history_to_limit( + &self, + max_entries: NonZeroU32, + ) -> Result { + let result = sqlx::query( + r" + delete from fingerprint_update_history as target + using ( + select id + from ( + select + id, + row_number() over (order by created_at desc, id desc) as position + from fingerprint_update_history + ) as ranked + where position > $1 + ) as expired + where target.id = expired.id + ", + ) + .bind(i64::from(max_entries.get())) + .execute(&self.pool) + .await?; + Ok(result.rows_affected()) + } } fn fingerprint_from_row(row: &sqlx::postgres::PgRow) -> Result { diff --git a/backend/tests/api/admin/settings_routes/routes.rs b/backend/tests/api/admin/settings_routes/routes.rs index fa8d18f37..9aec99a23 100644 --- a/backend/tests/api/admin/settings_routes/routes.rs +++ b/backend/tests/api/admin/settings_routes/routes.rs @@ -541,6 +541,6 @@ fn test_config(database_url: String) -> AppConfig { config.auth.refresh_margin_seconds = 240; config.auth.max_concurrent_per_account = 4; config.auth.tier_priority = vec!["team".to_string(), "plus".to_string()]; - config.logging.enabled = true; + config.telemetry.enabled = true; config } diff --git a/backend/tests/api/admin/usage_routes.rs b/backend/tests/api/admin/usage_routes.rs index cee833b36..63968f3d1 100644 --- a/backend/tests/api/admin/usage_routes.rs +++ b/backend/tests/api/admin/usage_routes.rs @@ -129,7 +129,7 @@ async fn monitoring_test_app( seed_admin_session(&pool, &redis, "session_1").await; } let mut config = test_config(test_database_url()); - config.logging.enabled = true; + config.telemetry.enabled = true; let stores = background_task_stores(pool.clone(), redis); let services = Services::new( &config, diff --git a/backend/tests/api/middleware/trace.rs b/backend/tests/api/middleware/trace.rs index 495eafacf..cdca789ed 100644 --- a/backend/tests/api/middleware/trace.rs +++ b/backend/tests/api/middleware/trace.rs @@ -17,6 +17,7 @@ use codex_proxy_rs::api::middleware::{ request_id::{attach_request_id, ClientIp}, trace::http_trace_layer, }; +use serde_json::Value; use tokio::time::{sleep, Duration}; use tower::ServiceExt; use tracing_subscriber::fmt::MakeWriter; @@ -74,6 +75,7 @@ async fn http_trace_should_include_request_id_and_completion_fields() { let subscriber = tracing_subscriber::fmt() .json() .with_writer(logs.clone()) + .with_max_level(tracing::Level::DEBUG) .with_current_span(true) .with_span_list(true) .with_ansi(false) @@ -109,6 +111,117 @@ async fn http_trace_should_include_request_id_and_completion_fields() { && output.contains("latency_ms"), "unexpected trace output: {output}" ); + let received = json_events(&output) + .into_iter() + .find(|event| event["fields"]["message"] == "received HTTP request") + .expect("received event should exist"); + assert!(received["fields"].get("request_id").is_none()); + assert!(received["fields"].get("method").is_none()); + assert!(received["fields"].get("uri").is_none()); +} + +#[tokio::test(flavor = "current_thread")] +async fn http_trace_should_include_resolved_client_ip_in_span() { + let logs = SharedLogBuffer::default(); + let subscriber = tracing_subscriber::fmt() + .json() + .with_writer(logs.clone()) + .with_current_span(true) + .with_span_list(true) + .with_ansi(false) + .finish(); + let app = Router::new() + .route("/trace-test", get(|| async { StatusCode::NO_CONTENT })) + .layer(http_trace_layer()) + .layer(from_fn(attach_request_id)); + let mut request = Request::builder() + .uri("/trace-test") + .header("cf-connecting-ip", "203.0.113.43") + .body(Body::empty()) + .unwrap(); + request + .extensions_mut() + .insert(ConnectInfo(SocketAddr::from(([127, 0, 0, 1], 61234)))); + + let _guard = tracing::subscriber::set_default(subscriber); + let response = app.oneshot(request).await.unwrap(); + let _body = to_bytes(response.into_body(), 1024).await.unwrap(); + + let output = wait_for_trace_output(&logs).await; + assert!( + output.contains("\"client_ip\":\"203.0.113.43\""), + "unexpected trace output: {output}" + ); +} + +#[tokio::test(flavor = "current_thread")] +async fn http_trace_should_emit_one_terminal_event_for_server_error() { + let logs = SharedLogBuffer::default(); + let subscriber = tracing_subscriber::fmt() + .json() + .with_writer(logs.clone()) + .with_current_span(true) + .with_span_list(true) + .with_ansi(false) + .finish(); + let app = Router::new() + .route( + "/trace-test", + get(|| async { StatusCode::INTERNAL_SERVER_ERROR }), + ) + .layer(http_trace_layer()) + .layer(from_fn(attach_request_id)); + + let _guard = tracing::subscriber::set_default(subscriber); + let response = app + .oneshot( + Request::builder() + .uri("/trace-test") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let _body = to_bytes(response.into_body(), 1024).await.unwrap(); + sleep(Duration::from_millis(10)).await; + + let output = logs.content(); + assert_eq!(output.matches("completed HTTP request").count(), 0); + assert_eq!(output.matches("failed HTTP request").count(), 1); +} + +#[tokio::test(flavor = "current_thread")] +async fn health_route_outside_trace_layer_should_not_emit_http_events() { + let logs = SharedLogBuffer::default(); + let subscriber = tracing_subscriber::fmt() + .json() + .with_writer(logs.clone()) + .with_current_span(true) + .with_span_list(true) + .with_ansi(false) + .finish(); + let traced_routes = Router::new() + .route("/trace-test", get(|| async { StatusCode::NO_CONTENT })) + .layer(http_trace_layer()); + let app = Router::new() + .route("/healthz", get(|| async { StatusCode::NO_CONTENT })) + .merge(traced_routes) + .layer(from_fn(attach_request_id)); + + let _guard = tracing::subscriber::set_default(subscriber); + let response = app + .oneshot( + Request::builder() + .uri("/healthz") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let _body = to_bytes(response.into_body(), 1024).await.unwrap(); + sleep(Duration::from_millis(10)).await; + + assert!(logs.content().is_empty()); } #[tokio::test(flavor = "current_thread")] @@ -218,6 +331,13 @@ async fn request_context_should_generate_new_request_id_for_invalid_header() { ); } +fn json_events(output: &str) -> Vec { + output + .lines() + .filter_map(|line| serde_json::from_str(line).ok()) + .collect() +} + #[tokio::test] async fn request_context_should_attach_client_ip_from_connection() { let app = Router::new() diff --git a/backend/tests/bootstrap/config/mod.rs b/backend/tests/bootstrap/config/mod.rs index 25be2324d..9caa806ae 100644 --- a/backend/tests/bootstrap/config/mod.rs +++ b/backend/tests/bootstrap/config/mod.rs @@ -72,8 +72,15 @@ admin: default_username: admin default_password: test-admin-password logging: - directory: .runtime/logs - retention_days: 14 + level: info + stdout: true + file: + enabled: true + directory: .runtime/logs + retention_days: 14 + max_file_size_mb: 20 + max_files: 20 +telemetry: enabled: true "#; @@ -109,7 +116,8 @@ fn default_config_keeps_runtime_artifacts_under_runtime_directory() { "postgres://codex_proxy:codex_proxy@127.0.0.1:5432/codex_proxy" ); assert_eq!(cfg.redis.url, "redis://127.0.0.1:6379"); - assert_eq!(cfg.logging.directory, ".runtime/logs"); + assert_eq!(cfg.logging.file.directory, ".runtime/logs"); + assert!(cfg.telemetry.enabled); } #[test] @@ -239,9 +247,14 @@ fingerprint: admin: session_ttl_minutes: 1440 logging: - directory: .runtime/logs - retention_days: 14 - enabled: false + level: info + stdout: true + file: + enabled: true + directory: .runtime/logs + retention_days: 14 + max_file_size_mb: 20 + max_files: 20 "#, ) .unwrap(); @@ -281,8 +294,10 @@ ws_pool: assert!(cfg.ws_pool.enabled); assert_eq!(cfg.ws_pool.max_age_ms, 3_300_000); assert_eq!(cfg.ws_pool.max_per_account, 8); - assert_eq!(cfg.logging.directory, ".runtime/logs"); - assert_eq!(cfg.logging.retention_days, 14); + assert_eq!(cfg.logging.file.directory, ".runtime/logs"); + assert_eq!(cfg.logging.file.retention_days, 14); + assert_eq!(cfg.logging.file.max_file_size_mb, 20); + assert_eq!(cfg.logging.file.max_files, 20); } #[test] @@ -315,17 +330,22 @@ fn config_loader_should_read_explicit_config_file_from_env() { std::env::remove_var("CPR_CONFIG_FILE"); assert_eq!(cfg.server.host, "127.0.0.2"); - assert_eq!(cfg.logging.directory, ".runtime/env-logs"); + assert_eq!(cfg.logging.file.directory, ".runtime/env-logs"); } #[test] fn logging_config_should_reject_unknown_fields() { let err = parse_yaml_config::( r" -directory: .runtime/logs +level: info +stdout: true unexpected: true -retention_days: 14 -enabled: false +file: + enabled: false + directory: .runtime/logs + retention_days: 14 + max_file_size_mb: 20 + max_files: 20 ", ) .unwrap_err(); diff --git a/backend/tests/dispatch/service.rs b/backend/tests/dispatch/service.rs index afa953b7e..577ef4283 100644 --- a/backend/tests/dispatch/service.rs +++ b/backend/tests/dispatch/service.rs @@ -363,7 +363,7 @@ async fn test_app_without_accounts( (router::router().with_state(state), api_key, pool, dir) } -async fn test_app_with_account_pool_and_logging( +async fn test_app_with_account_pool_and_telemetry( base_url: String, ) -> ( axum::Router, @@ -372,12 +372,12 @@ async fn test_app_with_account_pool_and_logging( crate::support::storage::TestDatabaseGuard, ) { test_app_with_account_pool_config(base_url, |config| { - config.logging.enabled = true; + config.telemetry.enabled = true; }) .await } -async fn test_app_with_account_pool_and_disabled_logging( +async fn test_app_with_account_pool_and_disabled_telemetry( base_url: String, ) -> ( axum::Router, @@ -403,7 +403,7 @@ async fn test_app_with_account_pool_and_disabled_logging( (router::router().with_state(state), api_key, pool, dir) } -async fn test_app_with_account_pool_and_logging_capture_body( +async fn test_app_with_account_pool_and_telemetry_capture_body( base_url: String, ) -> ( axum::Router, @@ -415,7 +415,7 @@ async fn test_app_with_account_pool_and_logging_capture_body( let api_key = insert_client_api_key(&pool).await; insert_account(&pool).await; let mut config = test_config(test_database_url(), base_url); - config.logging.enabled = true; + config.telemetry.enabled = true; let usage_record_options = UsageRecordOptions { enabled: true, capture_body: true, @@ -607,7 +607,7 @@ async fn test_app_with_two_accounts( (app, api_key, pool, dir) } -async fn test_app_with_two_accounts_and_logging( +async fn test_app_with_two_accounts_and_telemetry( base_url: String, ) -> ( axum::Router, @@ -617,7 +617,7 @@ async fn test_app_with_two_accounts_and_logging( ) { let (app, _state, api_key, pool, dir) = test_app_with_two_accounts_and_state_config(base_url, |config| { - config.logging.enabled = true; + config.telemetry.enabled = true; }) .await; (app, api_key, pool, dir) diff --git a/backend/tests/dispatch/service/compact_routes.rs b/backend/tests/dispatch/service/compact_routes.rs index cb27f8f79..c516b073a 100644 --- a/backend/tests/dispatch/service/compact_routes.rs +++ b/backend/tests/dispatch/service/compact_routes.rs @@ -64,7 +64,7 @@ async fn responses_review_route_should_record_review_route_in_usage_record() { .mount(&server) .await; - let (app, api_key, pool, _dir) = test_app_with_account_pool_and_logging(server.uri()).await; + let (app, api_key, pool, _dir) = test_app_with_account_pool_and_telemetry(server.uri()).await; let response = app .oneshot( Request::builder() @@ -121,7 +121,7 @@ async fn responses_compact_should_post_json_to_codex_compact_upstream() { .mount(&server) .await; - let (app, api_key, pool, _dir) = test_app_with_account_pool_and_logging(server.uri()).await; + let (app, api_key, pool, _dir) = test_app_with_account_pool_and_telemetry(server.uri()).await; let response = app .oneshot( Request::builder() @@ -243,7 +243,7 @@ async fn responses_compact_should_return_rate_limit_error_when_fallback_is_exhau .mount(&server) .await; - let (app, api_key, pool, _dir) = test_app_with_account_pool_and_logging(server.uri()).await; + let (app, api_key, pool, _dir) = test_app_with_account_pool_and_telemetry(server.uri()).await; let response = app .oneshot( Request::builder() @@ -309,7 +309,7 @@ async fn responses_compact_should_preserve_upstream_client_error_status() { .mount(&server) .await; - let (app, api_key, pool, _dir) = test_app_with_account_pool_and_logging(server.uri()).await; + let (app, api_key, pool, _dir) = test_app_with_account_pool_and_telemetry(server.uri()).await; let response = app .oneshot( Request::builder() diff --git a/backend/tests/dispatch/service/responses_http.rs b/backend/tests/dispatch/service/responses_http.rs index 2cec338bf..fff4f1aa0 100644 --- a/backend/tests/dispatch/service/responses_http.rs +++ b/backend/tests/dispatch/service/responses_http.rs @@ -267,7 +267,7 @@ async fn responses_should_use_websocket_upstream_by_default_while_serving_sse() async fn responses_non_stream_should_record_websocket_transport_metadata() { let (base_url, upstream) = spawn_single_websocket_completed_upstream("resp_ws_non_stream_log").await; - let (app, api_key, pool, _dir) = test_app_with_account_pool_and_logging(base_url).await; + let (app, api_key, pool, _dir) = test_app_with_account_pool_and_telemetry(base_url).await; let response = app .oneshot( diff --git a/backend/tests/dispatch/service/responses_recovery.rs b/backend/tests/dispatch/service/responses_recovery.rs index 7804a7af1..570cc2b7c 100644 --- a/backend/tests/dispatch/service/responses_recovery.rs +++ b/backend/tests/dispatch/service/responses_recovery.rs @@ -2081,7 +2081,7 @@ async fn responses_should_return_rate_limit_error_when_429_fallback_is_exhausted .mount(&server) .await; - let (app, api_key, pool, _dir) = test_app_with_account_pool_and_logging(server.uri()).await; + let (app, api_key, pool, _dir) = test_app_with_account_pool_and_telemetry(server.uri()).await; let response = app .oneshot( Request::builder() @@ -2153,7 +2153,7 @@ async fn responses_stream_should_return_rate_limit_error_when_429_fallback_is_ex .mount(&server) .await; - let (app, api_key, pool, _dir) = test_app_with_account_pool_and_logging(server.uri()).await; + let (app, api_key, pool, _dir) = test_app_with_account_pool_and_telemetry(server.uri()).await; let response = app .oneshot( Request::builder() diff --git a/backend/tests/dispatch/service/responses_websocket.rs b/backend/tests/dispatch/service/responses_websocket.rs index e544cd923..d388a9cd0 100644 --- a/backend/tests/dispatch/service/responses_websocket.rs +++ b/backend/tests/dispatch/service/responses_websocket.rs @@ -2313,7 +2313,7 @@ async fn responses_stream_with_previous_response_id_should_forward_websocket_chu async fn responses_stream_with_previous_response_id_should_record_websocket_audit_metadata() { let (base_url, first_chunk_sent_rx, finish_tx, upstream) = spawn_chunked_websocket_upstream().await; - let (app, api_key, pool, _dir) = test_app_with_account_pool_and_logging(base_url).await; + let (app, api_key, pool, _dir) = test_app_with_account_pool_and_telemetry(base_url).await; let response = app .oneshot(responses_previous_stream_request( diff --git a/backend/tests/dispatch/service/usage_logging.rs b/backend/tests/dispatch/service/usage_logging.rs index 864720e9a..69c4cf887 100644 --- a/backend/tests/dispatch/service/usage_logging.rs +++ b/backend/tests/dispatch/service/usage_logging.rs @@ -20,7 +20,7 @@ async fn responses_should_use_imported_account_record_usage_cookie_and_usage_rec .mount(&server) .await; - let (app, api_key, pool, _dir) = test_app_with_account_pool_and_logging(server.uri()).await; + let (app, api_key, pool, _dir) = test_app_with_account_pool_and_telemetry(server.uri()).await; let response = app .clone() .oneshot( @@ -101,7 +101,7 @@ async fn responses_usage_record_should_store_resolved_upstream_model_after_alias .await; let (app, api_key, pool, _dir) = test_app_with_account_pool_config(server.uri(), |config| { - config.logging.enabled = true; + config.telemetry.enabled = true; config .model_aliases .insert("client-alias".to_string(), "gpt-5.5".to_string()); @@ -143,7 +143,7 @@ async fn responses_usage_record_should_store_resolved_upstream_model_after_alias } #[tokio::test] -async fn responses_should_skip_usage_record_when_logging_disabled() { +async fn responses_should_skip_usage_record_when_telemetry_disabled() { let server = MockServer::start().await; Mock::given(method("POST")) .and(path("/codex/responses")) @@ -157,7 +157,7 @@ async fn responses_should_skip_usage_record_when_logging_disabled() { .await; let (app, api_key, pool, _dir) = - test_app_with_account_pool_and_disabled_logging(server.uri()).await; + test_app_with_account_pool_and_disabled_telemetry(server.uri()).await; let response = app .oneshot( Request::builder() @@ -534,7 +534,7 @@ async fn responses_stream_should_record_usage_record_after_completed_stream() { .mount(&server) .await; - let (app, api_key, pool, _dir) = test_app_with_account_pool_and_logging(server.uri()).await; + let (app, api_key, pool, _dir) = test_app_with_account_pool_and_telemetry(server.uri()).await; let response = app .oneshot( Request::builder() @@ -606,7 +606,7 @@ async fn responses_stream_should_preserve_body_metadata_when_capture_body_enable .await; let (app, api_key, pool, _dir) = - test_app_with_account_pool_and_logging_capture_body(server.uri()).await; + test_app_with_account_pool_and_telemetry_capture_body(server.uri()).await; let response = app .oneshot( Request::builder() @@ -657,7 +657,7 @@ async fn responses_stream_should_record_usage_record_after_late_disconnect() { ) .await; - let (app, api_key, pool, _dir) = test_app_with_account_pool_and_logging(base_url).await; + let (app, api_key, pool, _dir) = test_app_with_account_pool_and_telemetry(base_url).await; let response_task = tokio::spawn(async move { app.oneshot( Request::builder() @@ -723,7 +723,7 @@ async fn responses_stream_should_record_failure_detail_after_transport_error() { )) .await; - let (app, api_key, pool, _dir) = test_app_with_account_pool_and_logging(base_url).await; + let (app, api_key, pool, _dir) = test_app_with_account_pool_and_telemetry(base_url).await; let response_task = tokio::spawn(async move { app.oneshot( Request::builder() @@ -767,7 +767,7 @@ async fn responses_stream_should_not_record_first_token_when_metadata_only_prefi ) .await; - let (app, api_key, pool, _dir) = test_app_with_account_pool_and_logging(base_url).await; + let (app, api_key, pool, _dir) = test_app_with_account_pool_and_telemetry(base_url).await; let response_task = tokio::spawn(async move { app.oneshot( Request::builder() @@ -890,7 +890,7 @@ async fn responses_should_record_attempt_trace_after_retrying_another_account() .await; let secret_input = "do not persist this input"; - let (app, api_key, pool, _dir) = test_app_with_two_accounts_and_logging(server.uri()).await; + let (app, api_key, pool, _dir) = test_app_with_two_accounts_and_telemetry(server.uri()).await; let response = app .oneshot( Request::builder() diff --git a/backend/tests/infra/logging.rs b/backend/tests/infra/logging.rs index 1e33badfe..b13557289 100644 --- a/backend/tests/infra/logging.rs +++ b/backend/tests/infra/logging.rs @@ -1,6 +1,6 @@ use std::{fs, io::Write}; -use chrono::Utc; +use chrono::{Duration, Utc}; use codex_proxy_rs::infra::{ logging::{build_file_appender, RotationConfig}, time::china_date, @@ -9,18 +9,14 @@ use codex_proxy_rs::infra::{ #[test] fn rolling_appender_should_write_daily_codex_log_file() { let dir = tempfile::tempdir().unwrap(); - let config = RotationConfig::new(dir.path(), 14); + let config = rotation_config(dir.path(), 14, 1024, 20); let mut appender = build_file_appender(&config).unwrap(); appender.write_all(b"hello\n").unwrap(); appender.flush().unwrap(); - let names = fs::read_dir(dir.path()) - .unwrap() - .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned()) - .collect::>(); - - let expected = format!("codex-proxy-rs.{}.log", china_date(&Utc::now())); + let names = log_names(dir.path()); + let expected = managed_log_name(&china_date(&Utc::now()), 0); assert!( names.iter().any(|name| name == &expected), "expected China-date codex log file {expected}, found {names:?}" @@ -28,27 +24,101 @@ fn rolling_appender_should_write_daily_codex_log_file() { } #[test] -fn rotation_config_should_not_accept_size_limit() { - let config = RotationConfig::new("logs", 14); +fn rolling_appender_should_rotate_when_size_limit_is_exceeded() { + let dir = tempfile::tempdir().unwrap(); + let config = rotation_config(dir.path(), 14, 5, 20); + let mut appender = build_file_appender(&config).unwrap(); - assert_eq!(config.retention_days, 14); + appender.write_all(b"12345").unwrap(); + appender.write_all(b"6").unwrap(); + appender.flush().unwrap(); + + let date = china_date(&Utc::now()); + assert_eq!( + fs::read(dir.path().join(managed_log_name(&date, 0))).unwrap(), + b"12345" + ); + assert_eq!( + fs::read(dir.path().join(managed_log_name(&date, 1))).unwrap(), + b"6" + ); } #[test] -fn rolling_appender_should_remove_managed_logs_outside_retention() { +fn rolling_appender_should_remove_logs_outside_calendar_retention() { let dir = tempfile::tempdir().unwrap(); - fs::write(dir.path().join(managed_log_name("2000-01-01")), b"old").unwrap(); - fs::write(dir.path().join(managed_log_name("2001-01-01")), b"recent").unwrap(); + let old_date = china_date(&(Utc::now() - Duration::days(14))); + let retained_date = china_date(&(Utc::now() - Duration::days(13))); + fs::write(dir.path().join(managed_log_name(&old_date, 0)), b"old").unwrap(); + fs::write( + dir.path().join(managed_log_name(&retained_date, 0)), + b"retained", + ) + .unwrap(); fs::write(dir.path().join("other.log"), b"unmanaged").unwrap(); - let config = RotationConfig::new(dir.path(), 2); + let config = rotation_config(dir.path(), 14, 1024, 20); let _appender = build_file_appender(&config).unwrap(); - assert!(!dir.path().join(managed_log_name("2000-01-01")).exists()); - assert!(dir.path().join(managed_log_name("2001-01-01")).exists()); + assert!(!dir.path().join(managed_log_name(&old_date, 0)).exists()); + assert!(dir + .path() + .join(managed_log_name(&retained_date, 0)) + .exists()); assert!(dir.path().join("other.log").exists()); } -fn managed_log_name(date: &str) -> String { - format!("codex-proxy-rs.{date}.log") +#[test] +fn rolling_appender_should_enforce_global_file_count_limit() { + let dir = tempfile::tempdir().unwrap(); + let date = china_date(&Utc::now()); + for segment in 0..3 { + fs::write( + dir.path().join(managed_log_name(&date, segment)), + segment.to_string(), + ) + .unwrap(); + } + let config = rotation_config(dir.path(), 14, 1024, 2); + + let _appender = build_file_appender(&config).unwrap(); + + let names = log_names(dir.path()); + assert_eq!(names.len(), 2); + assert!(names.contains(&managed_log_name(&date, 2))); + assert!(names.contains(&managed_log_name(&date, 1))); +} + +#[test] +fn rolling_appender_should_reject_zero_limits() { + let dir = tempfile::tempdir().unwrap(); + + assert!(build_file_appender(&rotation_config(dir.path(), 0, 1024, 20)).is_err()); + assert!(build_file_appender(&rotation_config(dir.path(), 14, 0, 20)).is_err()); + assert!(build_file_appender(&rotation_config(dir.path(), 14, 1024, 0)).is_err()); +} + +fn rotation_config( + directory: &std::path::Path, + retention_days: usize, + max_file_size_bytes: u64, + max_files: usize, +) -> RotationConfig { + RotationConfig::new(directory, retention_days, max_file_size_bytes, max_files) +} + +fn log_names(directory: &std::path::Path) -> Vec { + fs::read_dir(directory) + .unwrap() + .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned()) + .filter(|name| name.starts_with("codex-proxy-rs.")) + .collect() +} + +fn managed_log_name(date: &str, segment: usize) -> String { + if segment == 0 { + format!("codex-proxy-rs.{date}.log") + } else { + format!("codex-proxy-rs.{date}.{segment}.log") + } } diff --git a/backend/tests/support/config.rs b/backend/tests/support/config.rs index 62cbf4218..94d213147 100644 --- a/backend/tests/support/config.rs +++ b/backend/tests/support/config.rs @@ -1,6 +1,7 @@ use codex_proxy_rs::bootstrap::config::{ - AdminConfig, ApiConfig, AppConfig, AuthConfig, DatabaseConfig, FingerprintConfig, - LoggingConfig, QuotaConfig, RedisConfig, ServerConfig, TlsConfig, WebSocketPoolSettings, + AdminConfig, ApiConfig, AppConfig, AuthConfig, DatabaseConfig, FileLoggingConfig, + FingerprintConfig, LoggingConfig, QuotaConfig, RedisConfig, ServerConfig, TelemetryConfig, + TlsConfig, WebSocketPoolSettings, }; pub(crate) fn test_config(database_url: String) -> AppConfig { @@ -43,9 +44,16 @@ pub(crate) fn test_config(database_url: String) -> AppConfig { default_password: "test-admin-password".to_string(), }, logging: LoggingConfig { - directory: "logs".to_string(), - retention_days: 14, - enabled: false, + level: "info".to_string(), + stdout: false, + file: FileLoggingConfig { + enabled: false, + directory: "logs".to_string(), + retention_days: 14, + max_file_size_mb: 20, + max_files: 20, + }, }, + telemetry: TelemetryConfig { enabled: false }, } } diff --git a/backend/tests/telemetry/account_usage/mod.rs b/backend/tests/telemetry/account_usage/mod.rs new file mode 100644 index 000000000..3d6074864 --- /dev/null +++ b/backend/tests/telemetry/account_usage/mod.rs @@ -0,0 +1 @@ +mod store; diff --git a/backend/tests/telemetry/account_usage/store.rs b/backend/tests/telemetry/account_usage/store.rs new file mode 100644 index 000000000..104668e0c --- /dev/null +++ b/backend/tests/telemetry/account_usage/store.rs @@ -0,0 +1,49 @@ +use std::num::NonZeroU32; + +use chrono::Utc; +use codex_proxy_rs::telemetry::account_usage::store::PgAccountUsageStore; + +use crate::support::storage::init_test_db; + +#[tokio::test] +async fn trim_model_usage_should_keep_latest_rows_per_account() { + let (pool, _guard) = init_test_db("account-model-usage-limit").await; + sqlx::query( + "insert into accounts (id, access_token, status, added_at, updated_at) + values ('account-limit', 'token', 'active', $1, $1)", + ) + .bind(Utc::now()) + .execute(&pool) + .await + .expect("insert account"); + sqlx::query( + "insert into account_model_usage (account_id, model, request_count, last_used_at) + select + 'account-limit', + 'model-' || lpad(sequence::text, 3, '0'), + 1, + $1 + sequence * interval '1 second' + from generate_series(0, 101) as sequence", + ) + .bind(Utc::now()) + .execute(&pool) + .await + .expect("insert model usage rows"); + + let deleted = PgAccountUsageStore::new(pool.clone()) + .trim_model_usage_to_limit(NonZeroU32::new(100).unwrap()) + .await + .expect("trim model usage rows"); + let (remaining, oldest_remaining): (i64, i64) = sqlx::query_as( + "select + count(*), + count(*) filter (where model in ('model-000', 'model-001')) + from account_model_usage + where account_id = 'account-limit'", + ) + .fetch_one(&pool) + .await + .expect("count retained model usage rows"); + + assert_eq!((deleted, remaining, oldest_remaining), (2, 100, 0)); +} diff --git a/backend/tests/telemetry/mod.rs b/backend/tests/telemetry/mod.rs index 5acb4c7af..8e3795c64 100644 --- a/backend/tests/telemetry/mod.rs +++ b/backend/tests/telemetry/mod.rs @@ -1,3 +1,4 @@ +pub(crate) mod account_usage; pub(crate) mod ops; pub(crate) mod rebuild; pub(crate) mod usage; diff --git a/backend/tests/upstream/openai/fingerprint/mod.rs b/backend/tests/upstream/openai/fingerprint/mod.rs index 6449ba000..ddc5275f8 100644 --- a/backend/tests/upstream/openai/fingerprint/mod.rs +++ b/backend/tests/upstream/openai/fingerprint/mod.rs @@ -1,3 +1,5 @@ +use std::num::NonZeroU32; + use codex_proxy_rs::upstream::openai::fingerprint::{PgFingerprintStore, UpdateChecker}; use crate::support::storage::init_test_db; @@ -34,6 +36,48 @@ async fn fingerprint_store_should_update_current_record() { assert_eq!(stored.chromium_version, "147"); } +#[tokio::test] +async fn fingerprint_store_should_keep_latest_update_history_rows() { + let (pool, _guard) = init_test_db("fingerprint-history-limit").await; + let repo = PgFingerprintStore::new(pool.clone()); + repo.ensure_current_seed(&crate::support::fingerprint::test_fingerprint()) + .await + .expect("seed current fingerprint"); + sqlx::query( + "insert into fingerprint_update_history ( + id, current_fingerprint_id, app_version, build_number, source, created_at + ) + select + 'history-' || lpad(sequence::text, 3, '0'), + 'current', + '26.900.' || sequence::text, + sequence::text, + 'auto_update', + $1 + sequence * interval '1 second' + from generate_series(0, 101) as sequence", + ) + .bind(chrono::Utc::now()) + .execute(&pool) + .await + .expect("insert fingerprint history rows"); + + let deleted = repo + .trim_update_history_to_limit(NonZeroU32::new(100).unwrap()) + .await + .expect("trim fingerprint history rows"); + let (remaining, oldest_remaining): (i64, i64) = sqlx::query_as( + "select + count(*), + count(*) filter (where id in ('history-000', 'history-001')) + from fingerprint_update_history", + ) + .fetch_one(&pool) + .await + .expect("count retained fingerprint history rows"); + + assert_eq!((deleted, remaining, oldest_remaining), (2, 100, 0)); +} + #[tokio::test] async fn update_checker_should_report_available_update_from_appcast() { let server = wiremock::MockServer::start().await; diff --git a/deploy/README.md b/deploy/README.md index 53638d092..97355fffb 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -17,8 +17,16 @@ cp deploy/config.example.yaml .runtime/config.yaml - `.runtime/data` -> `/app/data` - `.runtime/logs` -> `/app/logs` +应用日志同时写入 `docker logs` 与 `.runtime/logs`。Compose 对应用、PostgreSQL、Redis +统一启用 `json-file` 的 `10m × 5` 轮转；应用文件日志还受配置中的自然日、单文件大小和文件总数限制。 + 复制后必须设置 `admin.default_password`。空值、长度不足 12 位或常见弱口令会被后端拒绝。 +Redis 默认密码为 `codex_proxy`，与 `deploy/config.example.yaml` 的连接串一致。生产部署通过 +`CPR_REDIS_PASSWORD` 覆盖后，必须同步更新 `.runtime/config.yaml` 中 URL 的密码（特殊字符需做 URL 编码）。 +本地 Rust 集成测试若复用该 Compose Redis，设置 +`CPR_TEST_REDIS_URL=redis://:codex_proxy@127.0.0.1:6379/`。 + 如果前面还有 Nginx、Caddy、Cloudflare Tunnel 等反向代理，确保反代正常传递 `CF-Connecting-IP`、`X-Real-IP` 或 `X-Forwarded-For`。后端会按这组头自动解析真实客户端 IP；没有这些头时回落到直连 peer IP。 构建并启动： diff --git a/deploy/config.example.yaml b/deploy/config.example.yaml index aacbb4672..4b1b50807 100644 --- a/deploy/config.example.yaml +++ b/deploy/config.example.yaml @@ -14,7 +14,7 @@ database: redis: # Redis 仅承载可丢弃的运行态与缓存。 - url: 'redis://redis:6379/' + url: 'redis://:codex_proxy@redis:6379/' quota: # 账号额度刷新间隔，单位分钟。 @@ -89,8 +89,18 @@ admin: default_password: '' logging: - # Docker 部署默认写入 /app/logs，由 Compose 挂载到仓库 .runtime/logs。 - directory: /app/logs - # 日志保留天数。 - retention_days: 14 + # RUST_LOG 存在时优先使用环境变量。 + level: info + # 同时写入 docker logs。 + stdout: true + file: + # 文件日志挂载到仓库 .runtime/logs。 + enabled: true + directory: /app/logs + retention_days: 14 + max_file_size_mb: 20 + max_files: 20 + +telemetry: + # 记录真实 /v1 代理调用的成功与失败事实。 enabled: true diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index 9b793ed5e..12ce7e183 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -1,5 +1,11 @@ name: codex-proxy-rs +x-json-logging: &json-logging + driver: json-file + options: + max-size: "10m" + max-file: "5" + services: postgres: image: postgres:18-alpine @@ -19,22 +25,39 @@ services: retries: 10 security_opt: - no-new-privileges:true + logging: *json-logging redis: image: redis:8-alpine restart: unless-stopped - command: ["redis-server", "--appendonly", "yes"] + command: + [ + "redis-server", + "--appendonly", + "yes", + "--requirepass", + "${CPR_REDIS_PASSWORD:-codex_proxy}", + ] volumes: - redis-data:/data ports: - "127.0.0.1:6379:6379" healthcheck: - test: ["CMD", "redis-cli", "ping"] + test: + [ + "CMD", + "redis-cli", + "--no-auth-warning", + "-a", + "${CPR_REDIS_PASSWORD:-codex_proxy}", + "ping", + ] interval: 5s timeout: 5s retries: 10 security_opt: - no-new-privileges:true + logging: *json-logging codex-proxy-rs: image: ${CPR_IMAGE:-ghcr.io/zyycn/codex-proxy-rs:latest} @@ -90,6 +113,7 @@ services: CPR_UPDATE_LOCK_FILE: /app/data/update.lock CPR_WEB_DIST_DIR: /app/web/dist CPR_ENABLE_SELF_RESTART: "true" + logging: *json-logging volumes: postgres-data: diff --git a/docs/database.md b/docs/database.md index 0b949ff7e..3f5540006 100644 --- a/docs/database.md +++ b/docs/database.md @@ -124,12 +124,17 @@ | `ops_error_logs` | 30 天(默认) | `runtime_settings.ops_error_retention_days`,同上 | | `request_time_buckets` | 90 天(默认) | `runtime_settings.bucket_retention_days`,同上 | | `account_cookies` | 按 `expires_at` 清理 | 周期任务(PG) | +| `account_usage` | 每账号 1 行 | 账号删除时 FK 级联删除 | +| `account_model_usage` | 每账号最近 100 个模型 | 每小时按 `last_used_at desc nulls last, model` 裁剪 | +| `fingerprints` | 当前指纹固定 1 行 | 固定 ID 覆盖更新 | +| `fingerprint_update_history` | 全局最近 100 行 | 每小时按 `created_at desc, id desc` 裁剪 | | Redis 会话/租约/亲和 | TTL 自然过期 | 无清理任务 | -| 其余 PG 表 | 常量级增长 | — | +| 其余 PG 实体表 | 随管理实体数量增长 | 不随请求数增长 | -- trim 是**周期后台任务**(每轮读 `runtime_settings` 三列),不再在每次写入后内联执行——v3 的"每写一次 delete 一次"是已知的写放大缺陷,终态修复。 +- trim 是**每小时执行的后台任务**:三张时间表每轮读 `runtime_settings` 保留列,两张历史/统计表使用上表的固定行数上限。不再在每次写入后内联执行——v3 的"每写一次 delete 一次"是已知的写放大缺陷,终态修复。 +- 30/30/90 天是数据时间窗,不是字节硬上限;高流量下时间窗内的事实仍可很大。PostgreSQL `autovacuum` 会回收 DELETE 后的页供后续复用,但不保证数据文件立即缩小;不周期执行会锁表的 `VACUUM FULL`。 - `request_time_buckets` 在事实保留期内可由两张事实表重算(`rebuild-buckets` 命令,§7);超出事实保留期的桶是**因事实过期而升格的一级数据**,写入纪律 + 同事务写入(§5.2)是它们的唯一保障。 -- `account_usage` / `account_model_usage` 的累计列**不可重建**(事实早已过期),同样只受写入纪律保护。本文档不声称"一切派生皆可重算"。 +- `account_usage` / `account_model_usage` 的累计列**不可重建**(事实早已过期),同样只受写入纪律保护。模型维度超过 100 后丢弃最久未用的分项累计,账号总累计仍由 `account_usage` 保留。本文档不声称"一切派生皆可重算"。 - `window_*` 列是运行态(窗口边界来自上游),不可重建,可丢弃。 --- @@ -379,6 +384,8 @@ create index idx_account_model_usage_last_used `error_count` 是**收窄口径**:仅统计已归属到账号+模型之后的失败,用作调度回避信号(频繁失败 → 降权)。这是全库唯一允许的口径重叠(与 ops_error_logs),因为用途不同(调度局部信号 vs 全局错误事实);Dashboard 错误率禁止取自本列。 +每小时按 `last_used_at desc nulls last, model` 为每个账号保留前 100 行。该硬上限限制历史模型名持续增长,不改变 `account_usage` 的账号总累计。 + ### 4.8 usage_records ★ **每个成功的客户端请求恰好一行**。用量、token、成本、账号与调用方归因的唯一事实来源。这是系统的账本;"只收成功"的边界由 DB 约束兜底,不再只靠 service 层。 @@ -589,6 +596,8 @@ create table fingerprints ( 指纹更新审计。版本快照列固化"当时更新到了什么"——不能只存 FK,指纹行会被后续更新覆写。级联删除可接受:指纹本体删除后其历史无独立价值。 +每小时按 `created_at desc, id desc` 全局保留最新 100 行,让审计深度确定且不随运行年限增长。 + ```sql create table fingerprint_update_history ( id text primary key, @@ -777,7 +786,7 @@ codex-proxy-rs import-sqlite <旧库路径.sqlite> 6. **查询路径**(`telemetry/{usage,ops,account_usage}`、`api/admin/{usage,ops,dashboard,accounts}_routes*`):summary/分布/趋势全部走列,删除所有 `->>` 聚合与 metadata LIKE;新增 `/api/admin/ops/errors` 错误明细查询面;Dashboard 按 §5.3 只改取数来源，不改既有文案。 7. **运行态改 Redis**(`auth/store.rs` 会话、`accounts/refresh/lease.rs`、`dispatch/affinity/store.rs`、`models/store.rs`):按 §4B 契约实现;删除 admin 会话/亲和清理任务与亲和重启恢复;账号删除路径挂 Redis 亲和级联(§4B.3)。 8. **维护命令**(`main.rs` CLI):`import-sqlite `(§6);`rebuild-buckets`——删除事实保留期内的桶并从两张事实表重算(15 分钟槽对齐在 SQL 侧 `floor(epoch/900)*900`,§2.3);保留期外只读不动。范围明确**不含** `account_usage` / `account_model_usage`(§2.10,不可重建)。 -9. **清理任务**(`bootstrap/tasks/`):三张增长表的周期 trim(读 `runtime_settings` 三列);cookie 清理保留;admin 会话与亲和清理任务删除(TTL 接管)。 +9. **清理任务**(`bootstrap/tasks/`):三张时间增长表的周期 trim(读 `runtime_settings` 三列),`account_model_usage` 每账号 100 行与 `fingerprint_update_history` 全局 100 行硬上限;cookie 清理保留;admin 会话与亲和清理任务删除(TTL 接管)。 10. **健康检查**(`api/router.rs`):PG `select 1` + Redis `PING`,任一失败即 503。 11. **测试**(`backend/tests/`):support 换 PG(每测试独立数据库,`CPR_TEST_DATABASE_URL` 指定服务端)+ Redis(每测试随机键前缀,`CPR_TEST_REDIS_URL`);storage 测试断言终态 DDL、导入拆分规则(成功行迁入 / error 行转 ops / 噪音行丢弃 / 旧 `refreshing` 规范化)、两条写入事务的原子性、0ms 延迟样本可入桶、租约原子性。 12. **部署**(`deploy/docker-compose.yml`):新增 postgres、redis 服务与健康检查依赖。 diff --git a/docs/deploy-ci-review.md b/docs/deploy-ci-review.md index 82d009929..e4f7b254f 100644 --- a/docs/deploy-ci-review.md +++ b/docs/deploy-ci-review.md @@ -21,12 +21,16 @@ ## P1 — Redis 裸奔,与 Postgres 不一致 +> 状态：2026-07-11 已完成。Compose 使用 `CPR_REDIS_PASSWORD` 配置 `requirepass`，健康检查与应用连接串已同步。 + - 现状:`docker-compose.yml` 中 Postgres 有密码,Redis 没有 `requirepass`——同 compose 网络内任何容器、以及宿主机本地进程(绑了 `127.0.0.1:6379`)均可无认证读写。 - 建议:`command` 加 `--requirepass ${CPR_REDIS_PASSWORD:-...}`,应用侧连接串同步。 - 顺带:pg / redis 的宿主端口映射(`127.0.0.1:5432` / `127.0.0.1:6379`)若只是偶尔调试用,建议注释掉——应用走 compose 内网,不需要它们。 ## P1 — 日志与资源限制不一致 +> 状态：2026-07-11 已完成日志上限；PostgreSQL / Redis 资源上限仍待单独评估。三服务统一使用 `json-file` `10m × 5`。 + - 现状:三个服务的 stdout 走默认 `json-file` driver,没配 `max-size` / `max-file`,长期运行吃满磁盘(容器内 `/app/logs` 还写一份);postgres / redis 没有任何资源限制,而 app 有(`cpus` / `mem_limit` / `pids_limit`)——PG 内存失控会拖垮宿主。 - 建议:三个服务统一加 `logging` 限制(或宿主 daemon.json 全局配);给 pg / redis 补资源上限;Postgres 大查询会碰默认 64MB shm,可按需加 `shm_size`。 diff --git a/docs/logging-review.md b/docs/logging-review.md index 87422c8a7..0f241700f 100644 --- a/docs/logging-review.md +++ b/docs/logging-review.md @@ -2,22 +2,22 @@ > 审查基线：`5150eb7a`，分支 `feat/postgres-redis-migration`。 > -> 状态：设计已确认，代码实施待开始。 +> 状态：2026-07-11 已完成代码实施、全量质量门禁与容器运行验收。 > > 范围：应用 tracing、HTTP 访问日志、Docker 容器日志、PostgreSQL stderr/WAL，以及 PostgreSQL 日志表与统计表的增长边界。 ## 1. 结论 -当前日志系统的结构化、脱敏和文件轮转基础是正确的，但有四个需要优先修复的问题： +审查基线的日志系统具备结构化、脱敏和按日文件轮转基础，但有四个需要优先修复的问题： 1. 应用只写文件，`docker logs` 为空；关闭文件日志会同时关闭全部 tracing。 2. 文件 writer 使用 `tracing_appender` 默认 lossy 模式，缓冲满时会静默丢日志，当前没有上报丢弃计数。 3. HTTP 请求日志字段重复，5xx 同时产生 completed 和 failed 两条终态事件，`/healthz` 每 30 秒产生两条无价值 INFO。 4. Docker 的应用、PostgreSQL、Redis 都使用无大小上限的 `json-file` 日志。 -PostgreSQL 三张主要增长表已有按时间保留策略，不会随运行年限无限增长；但是 `account_model_usage` 和 `fingerprint_update_history` 目前没有严格上限，需要补充硬上限。 +PostgreSQL 三张主要增长表已有按时间保留策略；基线中 `account_model_usage` 和 `fingerprint_update_history` 没有严格上限。本轮已关闭全部四项日志问题并补齐两张表的 100 行硬上限。 -## 2. 已核实的现状 +## 2. 已核实的改造前现状 ### 2.1 应用日志 @@ -77,9 +77,9 @@ PostgreSQL 三张主要增长表已有按时间保留策略，不会随运行年 | `request_time_buckets` | 默认 90 天 | 已按 `bucket_start` 删除，不会按运行年限无限增长 | | `account_cookies` | 按 `expires_at` 周期清理 | 有过期时间的 Cookie 有边界；会话 Cookie 受唯一键约束但没有时间保留期 | | `account_usage` | 每账号一行，账号删除级联 | 行数上限等于账号数，不随请求数增长 | -| `account_model_usage` | 每账号/模型一行 | 历史模型名可持续增加；新增每账号最多保留最近 100 个模型的硬上限 | +| `account_model_usage` | 基线为每账号/模型一行 | 已新增每账号最多保留最近 100 个模型的硬上限 | | `fingerprints` | 当前指纹固定 ID 覆盖更新 | 常量级 | -| `fingerprint_update_history` | 每次实际更新插入一行 | 当前无上限；新增全局最多保留最近 100 条的硬上限 | +| `fingerprint_update_history` | 基线为每次实际更新插入一行 | 已新增全局最多保留最近 100 条的硬上限 | 时间保留限制的是数据时间范围，不是固定字节数：高流量下 30 天事实仍可达到很大体量。当前 PostgreSQL `autovacuum=on`，会回收 DELETE 后的页面供后续复用，但不会保证数据文件立即缩小；不应周期执行会锁表的 `VACUUM FULL`。 @@ -160,13 +160,13 @@ logging: 本轮范围为全部 P0、P1，以及 panic hook 和启动期日志级别配置。运行时动态调级与故障风暴采样明确不在本轮范围。 -- [ ] P0：重构日志配置和 stdout/file 双输出。 -- [ ] P0：增加 Docker 三服务日志上限。 -- [ ] P0：保留并监测 non-blocking 丢弃计数。 -- [ ] P1：healthz 降噪、ClientIp span、HTTP 单终态事件。 -- [ ] P1：本地数据丢失路径升级为 ERROR。 -- [ ] P1：补齐 `account_model_usage` / `fingerprint_update_history` 硬上限。 -- [ ] P2：panic hook，保留原 hook 与 backtrace 行为。 +- [x] P0：重构日志配置和 stdout/file 双输出。 +- [x] P0：增加 Docker 三服务日志上限。 +- [x] P0：保留并监测 non-blocking 丢弃计数。 +- [x] P1：healthz 降噪、ClientIp span、HTTP 单终态事件。 +- [x] P1：本地数据丢失路径升级为 ERROR。 +- [x] P1：补齐 `account_model_usage` / `fingerprint_update_history` 硬上限。 +- [x] P2：panic hook，保留原 hook 与 backtrace 行为。 - [ ] P2：运行时动态调整日志级别；本轮先支持启动配置和 `RUST_LOG`。 - [ ] P2：上游故障风暴采样/限频；有实际告警系统后再设计。 @@ -181,4 +181,16 @@ logging: 7. 三个容器 `json-file` 均为 `10m × 5`。 8. 全量 retention 后，三张时间表满足保留期，两个历史/统计表满足 100 行硬上限。 9. PostgreSQL 维持 `logging_collector=off`、无归档、无复制槽时 WAL 正常回收。 -10. Rust fmt、check、clippy `-D warnings`、612 项测试、Compose config、镜像构建与干净启动全部通过。 +10. Rust fmt、check、clippy `-D warnings`、全量测试、Compose config、镜像构建与干净启动全部通过。 + +## 6. 验收结果（2026-07-11） + +- Rust：`cargo fmt --check`、all-targets/all-features Clippy `-D warnings` 全部通过；3 项库单测 + 619 项集成测试，0 失败。 +- 前端：锁定依赖安装、Prettier 检查、`vue-tsc` 和 Vite 生产构建通过。 +- 架构：权威文档 §9 临时门禁通过，没有恢复或留下检查脚本；所有 `backend/src` 文件不超过 800 行。 +- 日志：干净启动后 stdout 与文件各观测到同一批 10 条有效 JSON，WARN=0、ERROR=0；手工和 Docker 的 healthz 探测都不产生 HTTP trace。 +- 丢弃与轮转：小缓冲区单测确认紧急 JSON 告警；文件自然日、大小、总数上限以及 stdout-only/file-only 配置均通过测试。 +- 数据库：两项真实 PostgreSQL 测试均将 102 行精确裁剪为最新 100 行；本地演示数据的 8 账号、3 密钥、160 成功事实、28 错误事实、672 统计桶在换镜像前后数量一致。 +- PostgreSQL：`logging_collector=off`、`log_destination=stderr`、`archive_mode=off`、复制槽 0、WAL 实测 80MB；数据卷中没有 PG 日志文件。 +- Docker：镜像 `codex-proxy-rs:logging-review` 清单摘要 `sha256:cfdd38395d19602dbdd31aa1bae5a03eed2e071cf770b729849bd2db8605378f`；三服务均 healthy、restart=0、`json-file 10m × 5`，`/healthz` 返回 204。 +- 安全：Cargo audit、pnpm production audit 和镜像 HIGH/CRITICAL Trivy 均为 0 漏洞；Redis 已开启 `requirepass`，未认证 PING 返回 `NOAUTH`，启动 WARN=0。