From dfcd30bd8473e86c9fd0a444e2491825d13b6ee7 Mon Sep 17 00:00:00 2001 From: zyycn Date: Sun, 14 Jun 2026 02:51:35 +0800 Subject: [PATCH] refactor: modernize logging Co-authored-by: Codex --- config.yaml | 1 - docs/api.md | 2 +- docs/architecture-audit.md | 1 - .../plans/2026-06-14-logging-refactor.md | 68 ++++++ .../2026-06-14-logging-refactor-design.md | 51 ++++ src/admin/tasks/session_cleanup.rs | 11 +- src/codex/accounts/models/service.rs | 4 +- .../gateway/fingerprint/update_checker.rs | 29 ++- src/codex/gateway/installation.rs | 6 +- src/codex/gateway/transport/client.rs | 31 ++- src/codex/gateway/transport/websocket.rs | 21 ++ src/codex/logs/rotation.rs | 218 ++++-------------- src/codex/serving/chat.rs | 4 +- src/codex/serving/dispatch/mod.rs | 77 ++++--- src/codex/serving/dispatch/refresh.rs | 4 +- src/codex/serving/responses.rs | 36 +-- src/codex/tasks/model.rs | 24 +- src/codex/tasks/quota.rs | 19 +- src/codex/tasks/refresh.rs | 39 ++-- src/config/types.rs | 4 +- src/main.rs | 12 +- src/runtime/bootstrap.rs | 8 +- src/runtime/mod.rs | 2 +- src/runtime/router.rs | 102 +++++++- src/runtime/tasks/coordinator.rs | 22 +- tests/admin/api_keys_route.rs | 1 - tests/admin/login_route.rs | 1 - tests/admin/logs_route.rs | 1 - tests/admin/models_route.rs | 1 - tests/admin/settings_route.rs | 1 - tests/admin/usage_stats_route.rs | 1 - tests/codex_operations/log_rotation.rs | 42 ++-- tests/codex_serving/chat_completions.rs | 1 - tests/codex_serving/routes_responses.rs | 1 - tests/codex_serving/upstream_fallback.rs | 2 +- tests/config.rs | 25 +- tests/runtime.rs | 1 + tests/runtime/http_trace.rs | 93 ++++++++ tests/runtime/startup.rs | 1 - tests/support/admin_accounts.rs | 1 - tests/support/upstream.rs | 1 - 41 files changed, 603 insertions(+), 367 deletions(-) create mode 100644 docs/superpowers/plans/2026-06-14-logging-refactor.md create mode 100644 docs/superpowers/specs/2026-06-14-logging-refactor-design.md create mode 100644 tests/runtime/http_trace.rs diff --git a/config.yaml b/config.yaml index 08ddb192e..22acdcce6 100644 --- a/config.yaml +++ b/config.yaml @@ -41,7 +41,6 @@ admin: default_password: admin logging: directory: logs - max_file_bytes: 10485760 retention_days: 14 enabled: false capacity: 2000 diff --git a/docs/api.md b/docs/api.md index 0dab14f91..e80014ee3 100644 --- a/docs/api.md +++ b/docs/api.md @@ -334,7 +334,7 @@ Example item: "model": "gpt-5.5", "statusCode": 200, "latencyMs": 123, - "message": "v1 responses completed", + "message": "v1 responses 已完成", "metadata": { "stream": false, "usage": { diff --git a/docs/architecture-audit.md b/docs/architecture-audit.md index 41c3737e2..4c1888a5d 100644 --- a/docs/architecture-audit.md +++ b/docs/architecture-audit.md @@ -256,7 +256,6 @@ pub enum AccountServiceError { ```yaml logging: capacity: 2000 # SQLite 内存容量 - max_file_bytes: 10485760 # 10MB per file retention_days: 14 ``` diff --git a/docs/superpowers/plans/2026-06-14-logging-refactor.md b/docs/superpowers/plans/2026-06-14-logging-refactor.md new file mode 100644 index 000000000..347b598ce --- /dev/null +++ b/docs/superpowers/plans/2026-06-14-logging-refactor.md @@ -0,0 +1,68 @@ +# Logging Refactor Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Replace the custom file logging path with the Rust community `tracing-appender` stack and standardize source log messages in Chinese. + +**Architecture:** Keep two separate logging concerns. Runtime process logs go through `tracing` JSON output and daily rolling files. Admin queryable business events stay in SQLite through `EventLogRepository` and `LogService`. + +**Tech Stack:** Rust, `tracing`, `tracing-subscriber`, `tracing-appender`, `tower-http::trace`, `axum`, `serde_yml`, `thiserror`. + +--- + +### Task 1: Remove Size-Based Logging Configuration + +**Files:** +- Modify: `src/config/types.rs` +- Modify: `config.yaml` +- Modify: `tests/config.rs` +- Modify: test support configs under `tests/` + +- [x] Write a failing test that rejects `logging.max_file_bytes`. +- [x] Remove `max_file_bytes` from `LoggingConfig` and default config. +- [x] Update all test fixtures that construct `LoggingConfig`. +- [x] Run `cargo test --test config`. + +### Task 2: Replace Custom Rotating Writer + +**Files:** +- Modify: `src/codex/logs/rotation.rs` +- Modify: `tests/codex_operations/log_rotation.rs` +- Modify: `src/main.rs` + +- [x] Replace the old size/date writer with `RollingFileAppender::builder()`. +- [x] Return a guard that keeps the non-blocking writer alive. +- [x] Configure JSON subscriber output with target, level, file, line, current span, and span list. +- [x] Run `cargo test --test codex_operations`. + +### Task 3: Improve HTTP Trace Spans + +**Files:** +- Modify: `src/runtime/router.rs` +- Test: `tests/runtime/http_trace.rs` + +- [x] Add a test that verifies request logs include `request_id`, method, uri, status, and latency. +- [x] Build a custom `TraceLayer` that logs Chinese lifecycle messages and avoids headers/bodies. +- [x] Keep request id middleware in the correct layer order. +- [x] Run the HTTP trace test. + +### Task 4: Standardize Structured Logs + +**Files:** +- Modify: `src/**/*.rs` + +- [x] Convert string-formatted logs to structured fields. +- [x] Convert all source log messages to Chinese, preserving technical English terms where clearer. +- [x] Change expected WebSocket fallback from `warn` to `info`. +- [x] Add focused `#[tracing::instrument]` spans around upstream response transport paths without logging secrets. + +### Task 5: Verify and Commit + +**Files:** +- All changed files + +- [x] Run `cargo fmt --check`. +- [x] Run `git diff --check`. +- [x] Run `cargo test`. +- [x] Run `cargo clippy --all-targets --all-features --locked -- -D warnings`. +- [x] Commit with a short message and the `Co-authored-by: Codex ` trailer. diff --git a/docs/superpowers/specs/2026-06-14-logging-refactor-design.md b/docs/superpowers/specs/2026-06-14-logging-refactor-design.md new file mode 100644 index 000000000..685369e97 --- /dev/null +++ b/docs/superpowers/specs/2026-06-14-logging-refactor-design.md @@ -0,0 +1,51 @@ +# Logging Refactor Design + +**Date:** 2026-06-14 +**Status:** Approved + +## Goal + +Bring runtime logging in line with Rust community practice while the project is still in active refactor. The logging path should be simple, structured, safe to query, and free of compatibility code from the earlier custom rotation implementation. + +## Decisions + +- Use `tracing`, `tracing-subscriber`, `tracing-appender`, and `tower-http::trace::TraceLayer` as the runtime file logging stack. +- Remove the custom `RotatingLogWriter` and size-based `max_file_bytes` configuration. File logs rotate daily through `tracing-appender`, and retention is controlled by `retention_days`. +- Keep SQLite `EventLogRepository` and `LogService`. Those records power `/admin/logs` and business event queries; they are not the same concern as process file logs. +- HTTP tracing must create structured spans with `request_id`, `method`, and `uri`, then record completion with `status` and `latency_ms`. +- Log messages are written in Chinese. Technical terms such as `WebSocket`, `HTTP SSE`, `SQLite`, and `request_id` may remain in English when that is clearer. +- Structured field names remain English because they are machine-facing query keys for JSON logs and log systems such as Loki or Elasticsearch. +- Logs must not include access tokens, refresh tokens, cookies, full headers, or request/response bodies by default. + +## Runtime Shape + +`main.rs` initializes tracing once and keeps the non-blocking log guard alive for the whole process. The subscriber writes JSON lines to a daily rolling appender under `logging.directory`. + +Each HTTP request receives an `x-request-id` through the existing middleware. `TraceLayer` then attaches that request id to the request span and emits Chinese lifecycle messages without logging sensitive headers or bodies. + +Upstream Codex calls add focused spans around transport choices. Expected fallback, such as WebSocket being unavailable for a non-history request, is an `info` event. Recoverable persistence failures remain `warn`, and unrecoverable server failures remain `error`. + +## Configuration + +The retained logging configuration is: + +```yaml +logging: + directory: logs + retention_days: 14 + enabled: false + capacity: 2000 + capture_body: false +``` + +`retention_days` maps to the number of daily log files retained by `tracing-appender`. `enabled`, `capacity`, and `capture_body` continue to belong to the SQLite admin event log subsystem. + +`max_file_bytes` is removed instead of deprecated. This avoids a configuration field that appears to work but no longer maps to the community appender behavior. + +## Validation + +- Config loading rejects the removed `max_file_bytes` field. +- File logging uses the `codex-proxy-rs..log` naming produced by `tracing-appender`. +- HTTP trace output includes `request_id`, `method`, `uri`, `status`, and `latency_ms`. +- All source `tracing` messages are Chinese, with professional English terms preserved where appropriate. +- `cargo fmt --check`, `cargo test`, and `cargo clippy --all-targets --all-features --locked -- -D warnings` must pass before committing. diff --git a/src/admin/tasks/session_cleanup.rs b/src/admin/tasks/session_cleanup.rs index 5b83e7a71..f8865a61c 100644 --- a/src/admin/tasks/session_cleanup.rs +++ b/src/admin/tasks/session_cleanup.rs @@ -20,28 +20,25 @@ impl SessionCleanupScheduler { tokio::spawn(async move { let mut ticker = interval(Duration::from_secs(self.interval_secs)); - info!( - interval_secs = self.interval_secs, - "Session cleanup scheduler started" - ); + info!(interval_secs = self.interval_secs, "会话清理调度器已启动"); loop { tokio::select! { _ = ticker.tick() => { match self.cleanup_expired_sessions().await { Ok(count) if count > 0 => { - info!(count, "Cleaned up expired sessions"); + info!(count, "已清理过期会话"); } Ok(_) => { // 没有过期会话，不输出日志 } Err(e) => { - error!(error = %e, "Failed to clean up expired sessions"); + error!(error = %e, "清理过期会话失败"); } } } _ = shutdown_rx.recv() => { - info!("Session cleanup scheduler shut down gracefully"); + info!("会话清理调度器已关闭"); break; } } diff --git a/src/codex/accounts/models/service.rs b/src/codex/accounts/models/service.rs index 318e6b657..e89d5b294 100644 --- a/src/codex/accounts/models/service.rs +++ b/src/codex/accounts/models/service.rs @@ -72,7 +72,7 @@ impl ModelService { Ok(_) => ModelCatalog::from_config(&self.config.model), Err(error) => { // Cached model snapshots extend the catalog; fallback keeps proxy routes available. - tracing::warn!(?error, "failed to load cached model snapshots"); + tracing::warn!(error = %error, "加载缓存模型快照失败"); ModelCatalog::from_config(&self.config.model) } } @@ -138,7 +138,7 @@ impl ModelService { continue; } Err(error) => { - tracing::warn!(?error, plan_type, "failed to refresh backend models"); + tracing::warn!(error = %error, plan_type, "刷新后端模型失败"); result.failed_plans += 1; continue; } diff --git a/src/codex/gateway/fingerprint/update_checker.rs b/src/codex/gateway/fingerprint/update_checker.rs index 1b769cf19..713ea85fc 100644 --- a/src/codex/gateway/fingerprint/update_checker.rs +++ b/src/codex/gateway/fingerprint/update_checker.rs @@ -79,19 +79,19 @@ impl UpdateChecker { let mut ticker = interval(POLL_INTERVAL); info!( - "[UpdateChecker] 启动后台指纹版本检查器，间隔：{:?}", - POLL_INTERVAL + interval_secs = POLL_INTERVAL.as_secs(), + "UpdateChecker 后台 fingerprint 版本检查器已启动" ); // 立即执行首次检查 if let Err(e) = self.check_and_apply_update().await { - warn!("[UpdateChecker] 首次检查失败: {}", e); + warn!(error = %e, "UpdateChecker 首次检查失败"); } loop { ticker.tick().await; if let Err(e) = self.check_and_apply_update().await { - warn!("[UpdateChecker] 定期检查失败: {}", e); + warn!(error = %e, "UpdateChecker 定期检查失败"); } } }) @@ -101,8 +101,12 @@ impl UpdateChecker { let update_state = self.check_for_update().await?; if update_state.update_available { - let version = update_state.latest_version.as_ref().unwrap(); - let build = update_state.latest_build.as_ref().unwrap(); + let (Some(version), Some(build)) = ( + update_state.latest_version.as_ref(), + update_state.latest_build.as_ref(), + ) else { + return Ok(()); + }; let state = self.state.lock().await; let current_version = state.current_version.clone(); @@ -110,8 +114,11 @@ impl UpdateChecker { drop(state); info!( - "[UpdateChecker] *** 发现新版本: v{} (build {}) — 当前: v{} (build {})", - version, build, current_version, current_build + version = %version, + build = %build, + current_version = %current_version, + current_build = %current_build, + "UpdateChecker 发现新的 fingerprint 版本" ); self.apply_version_update(version, build).await?; @@ -120,7 +127,11 @@ impl UpdateChecker { state.current_version = version.clone(); state.current_build = build.clone(); - info!("[UpdateChecker] 已自动应用: v{} (build {})", version, build); + info!( + version = %version, + build = %build, + "UpdateChecker 已自动应用 fingerprint 版本" + ); } Ok(()) diff --git a/src/codex/gateway/installation.rs b/src/codex/gateway/installation.rs index 3f3460991..e9e12478c 100644 --- a/src/codex/gateway/installation.rs +++ b/src/codex/gateway/installation.rs @@ -48,7 +48,11 @@ fn resolve_installation_id(database_path: Option<&str>) -> String { // 生成并持久化新 UUID let generated = Uuid::new_v4().to_string(); if let Err(e) = persist_uuid(&our_file, &generated) { - tracing::warn!("Failed to persist installation_id to {:?}: {}", our_file, e); + tracing::warn!( + error = %e, + path = %our_file.display(), + "持久化 installation_id 失败" + ); } return generated; } diff --git a/src/codex/gateway/transport/client.rs b/src/codex/gateway/transport/client.rs index 7b98878ba..b8272ba75 100644 --- a/src/codex/gateway/transport/client.rs +++ b/src/codex/gateway/transport/client.rs @@ -129,6 +129,15 @@ impl CodexBackendClient { self } + #[tracing::instrument( + skip(self, request, context), + fields( + request_id = %context.request_id, + account_id = %context.account_id.unwrap_or("unknown"), + model = %request.model, + transport = ?transport_for_request(request), + ) + )] pub async fn create_response( &self, request: &CodexResponsesRequest, @@ -159,9 +168,9 @@ impl CodexBackendClient { if transport == CodexTransport::WebSocketPreferred && websocket_error_allows_http_sse_fallback(&error) => { - tracing::warn!( + tracing::info!( error = %error, - "Codex WebSocket unavailable; falling back to HTTP SSE" + "Codex WebSocket 不可用，降级到 HTTP SSE" ); } Err(error) => return Err(codex_websocket_error(error)), @@ -196,6 +205,15 @@ impl CodexBackendClient { }) } + #[tracing::instrument( + skip(self, request, context), + fields( + request_id = %context.request_id, + account_id = %context.account_id.unwrap_or("unknown"), + model = %request.model, + transport = "http_sse", + ) + )] pub async fn stream_response( &self, request: &CodexResponsesRequest, @@ -229,6 +247,15 @@ impl CodexBackendClient { }) } + #[tracing::instrument( + skip(self, request, context), + fields( + request_id = %context.request_id, + account_id = %context.account_id.unwrap_or("unknown"), + model = %request.model, + transport = "websocket", + ) + )] pub async fn websocket_stream_response( &self, request: &CodexResponsesRequest, diff --git a/src/codex/gateway/transport/websocket.rs b/src/codex/gateway/transport/websocket.rs index dabbed167..33761719e 100644 --- a/src/codex/gateway/transport/websocket.rs +++ b/src/codex/gateway/transport/websocket.rs @@ -621,6 +621,13 @@ pub fn http_sse_fallback_allowed(request: &CodexResponsesRequest) -> bool { ) } +#[tracing::instrument( + skip(base_url, request, headers), + fields( + model = %request.model, + has_previous_response_id = request.previous_response_id.is_some(), + ) +)] pub async fn create_response_via_websocket( base_url: &str, request: &CodexResponsesRequest, @@ -649,6 +656,13 @@ pub async fn create_response_via_websocket( }) } +#[tracing::instrument( + skip(base_url, request, headers), + fields( + model = %request.model, + has_previous_response_id = request.previous_response_id.is_some(), + ) +)] pub async fn create_response_via_websocket_stream( base_url: &str, request: &CodexResponsesRequest, @@ -657,6 +671,13 @@ pub async fn create_response_via_websocket_stream( create_response_via_websocket_stream_inner(base_url, request, headers, None).await } +#[tracing::instrument( + skip(base_url, request, headers, pool, pool_key), + fields( + model = %request.model, + has_previous_response_id = request.previous_response_id.is_some(), + ) +)] pub async fn create_response_via_websocket_stream_with_pool( base_url: &str, request: &CodexResponsesRequest, diff --git a/src/codex/logs/rotation.rs b/src/codex/logs/rotation.rs index 06202f7a6..de4ac1d49 100644 --- a/src/codex/logs/rotation.rs +++ b/src/codex/logs/rotation.rs @@ -1,29 +1,28 @@ use std::{ - fs::{self, File, OpenOptions}, - io::{self, Write}, + fs, io, path::{Path, PathBuf}, - sync::{Arc, Mutex, MutexGuard}, - time::{Duration, SystemTime}, }; -use chrono::Utc; use thiserror::Error; -use tracing_subscriber::{fmt::MakeWriter, EnvFilter}; +use tracing_appender::{ + non_blocking::WorkerGuard, + rolling::{RollingFileAppender, Rotation}, +}; +use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter}; const LOG_PREFIX: &str = "codex-proxy-rs"; +const LOG_SUFFIX: &str = "log"; #[derive(Debug, Clone)] pub struct RotationConfig { pub directory: PathBuf, - pub max_file_bytes: u64, - pub retention_days: u64, + pub retention_days: usize, } impl RotationConfig { - pub fn new(directory: impl AsRef, max_file_bytes: u64, retention_days: u64) -> Self { + pub fn new(directory: impl AsRef, retention_days: usize) -> Self { Self { directory: directory.as_ref().to_path_buf(), - max_file_bytes, retention_days, } } @@ -33,181 +32,54 @@ impl RotationConfig { pub enum LogError { #[error("log io error: {0}")] Io(#[from] io::Error), + #[error("log appender initialization failed: {0}")] + Appender(#[from] tracing_appender::rolling::InitError), #[error("global tracing subscriber is already initialized")] SubscriberAlreadyInitialized, } -#[derive(Clone)] -pub struct RotatingLogWriter { - inner: Arc>, +pub struct LogGuard { + _guard: WorkerGuard, } -impl RotatingLogWriter { - pub fn new(config: RotationConfig) -> Result { - fs::create_dir_all(&config.directory)?; - let current_date = current_date(); - let active_path = active_path(&config.directory, ¤t_date); - if active_path.exists() - && fs::metadata(&active_path)?.len() >= config.max_file_bytes - && config.max_file_bytes > 0 - { - rotate_existing(&active_path, ¤t_date)?; - } - cleanup_retention(&config.directory, config.retention_days)?; - let file = open_append(&active_path)?; - let current_size = file.metadata()?.len(); - Ok(Self { - inner: Arc::new(Mutex::new(RotatingLogInner { - config, - current_date, - current_size, - file, - })), - }) - } +pub fn build_file_appender(config: &RotationConfig) -> Result { + fs::create_dir_all(&config.directory)?; - pub fn active_path(&self) -> PathBuf { - let inner = self.lock_inner(); - active_path(&inner.config.directory, &inner.current_date) - } + let builder = RollingFileAppender::builder() + .rotation(Rotation::DAILY) + .filename_prefix(LOG_PREFIX) + .filename_suffix(LOG_SUFFIX); + let builder = if config.retention_days == 0 { + builder + } else { + builder.max_log_files(config.retention_days) + }; - fn lock_inner(&self) -> MutexGuard<'_, RotatingLogInner> { - match self.inner.lock() { - Ok(guard) => guard, - Err(poisoned) => poisoned.into_inner(), - } - } + builder.build(&config.directory).map_err(LogError::Appender) } -impl<'a> MakeWriter<'a> for RotatingLogWriter { - type Writer = RotatingLogFile; - - fn make_writer(&'a self) -> Self::Writer { - RotatingLogFile { - inner: Arc::clone(&self.inner), - } - } -} - -pub struct RotatingLogFile { - inner: Arc>, -} - -impl Write for RotatingLogFile { - fn write(&mut self, buf: &[u8]) -> io::Result { - let mut inner = match self.inner.lock() { - Ok(guard) => guard, - Err(poisoned) => poisoned.into_inner(), - }; - inner.roll_if_needed(buf.len() as u64)?; - let written = inner.file.write(buf)?; - inner.current_size += written as u64; - Ok(written) - } - - fn flush(&mut self) -> io::Result<()> { - let mut inner = match self.inner.lock() { - Ok(guard) => guard, - Err(poisoned) => poisoned.into_inner(), - }; - inner.file.flush() - } -} - -struct RotatingLogInner { - config: RotationConfig, - current_date: String, - current_size: u64, - file: File, -} - -impl RotatingLogInner { - fn roll_if_needed(&mut self, incoming_bytes: u64) -> io::Result<()> { - let today = current_date(); - let date_changed = today != self.current_date; - let size_exceeded = self.config.max_file_bytes > 0 - && self.current_size > 0 - && self.current_size.saturating_add(incoming_bytes) > self.config.max_file_bytes; - if !date_changed && !size_exceeded { - return Ok(()); - } - - self.file.flush()?; - let old_active_path = active_path(&self.config.directory, &self.current_date); - if old_active_path.exists() { - rotate_existing(&old_active_path, &self.current_date)?; - } - cleanup_retention(&self.config.directory, self.config.retention_days)?; - self.current_date = today; - let new_active_path = active_path(&self.config.directory, &self.current_date); - self.file = open_append(&new_active_path)?; - self.current_size = self.file.metadata()?.len(); - Ok(()) - } -} - -pub fn init_tracing(config: RotationConfig) -> Result { - let writer = RotatingLogWriter::new(config)?; +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")); - // 日志必须写入可轮转文件，不能只依赖 stdout，否则线上问题无法回溯。 - tracing_subscriber::fmt() - .json() - .with_env_filter(filter) - .with_writer(writer.clone()) + + tracing_subscriber::registry() + .with(filter) + .with( + fmt::layer() + .json() + .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), + ) .try_init() .map_err(|_| LogError::SubscriberAlreadyInitialized)?; - Ok(writer) -} -fn current_date() -> String { - Utc::now().format("%Y-%m-%d").to_string() -} - -fn active_path(directory: &Path, date: &str) -> PathBuf { - directory.join(format!("{LOG_PREFIX}.{date}.log")) -} - -fn archive_path(active_path: &Path, date: &str) -> PathBuf { - let stamp = Utc::now().format("%H%M%S%.3f"); - let suffix = uuid::Uuid::new_v4(); - active_path.with_file_name(format!("{LOG_PREFIX}.{date}.{stamp}.{suffix}.log")) -} - -fn rotate_existing(active_path: &Path, date: &str) -> io::Result<()> { - if fs::metadata(active_path) - .map(|metadata| metadata.len()) - .unwrap_or(0) - == 0 - { - return Ok(()); - } - fs::rename(active_path, archive_path(active_path, date)) -} - -fn open_append(path: &Path) -> io::Result { - OpenOptions::new().create(true).append(true).open(path) -} - -fn cleanup_retention(directory: &Path, retention_days: u64) -> io::Result<()> { - if retention_days == 0 { - return Ok(()); - } - let cutoff = SystemTime::now() - .checked_sub(Duration::from_secs(retention_days.saturating_mul(86_400))) - .unwrap_or(SystemTime::UNIX_EPOCH); - for entry in fs::read_dir(directory)? { - let entry = entry?; - let path = entry.path(); - let Some(file_name) = path.file_name().and_then(|value| value.to_str()) else { - continue; - }; - if !file_name.starts_with(LOG_PREFIX) || !file_name.ends_with(".log") { - continue; - } - let metadata = entry.metadata()?; - if metadata.modified().is_ok_and(|modified| modified < cutoff) { - let _ = fs::remove_file(path); - } - } - Ok(()) + Ok(LogGuard { _guard: guard }) } diff --git a/src/codex/serving/chat.rs b/src/codex/serving/chat.rs index bd819e574..d262730be 100644 --- a/src/codex/serving/chat.rs +++ b/src/codex/serving/chat.rs @@ -134,7 +134,7 @@ impl ChatService { &log_context, retry.status(), EventLevel::Warn, - "v1 chat completions upstream retrying with fallback account", + "v1 chat completions 上游请求将使用备用账户重试", retry.metadata(client_stream), ) .await; @@ -150,7 +150,7 @@ impl ChatService { &log_context, error_response.0, EventLevel::Error, - "v1 chat completions upstream request failed", + "v1 chat completions 上游请求失败", json!({"stream": client_stream}), ) .await; diff --git a/src/codex/serving/dispatch/mod.rs b/src/codex/serving/dispatch/mod.rs index f4d76ef50..74b0e4e1a 100644 --- a/src/codex/serving/dispatch/mod.rs +++ b/src/codex/serving/dispatch/mod.rs @@ -301,7 +301,7 @@ async fn responses_http_sse_stream( &log_context, retry.status(), EventLevel::Warn, - "v1 responses stream upstream retrying with fallback account", + "v1 responses stream 上游请求将使用备用账户重试", retry.metadata(true), ) .await; @@ -317,7 +317,7 @@ async fn responses_http_sse_stream( &log_context, error_response.0, EventLevel::Error, - "v1 responses stream upstream request failed", + "v1 responses stream 上游请求失败", json!({"stream": true}), ) .await; @@ -336,7 +336,7 @@ async fn responses_http_sse_stream( &log_context, StatusCode::INTERNAL_SERVER_ERROR, EventLevel::Error, - "v1 responses stream cookie persistence failed", + "v1 responses stream 持久化 cookie 失败", json!({"stream": true, "cookieStoreError": true}), ) .await; @@ -674,7 +674,7 @@ async fn record_response_affinity_with_deps( Err(error) => { tracing::warn!( error = %error, - "failed to parse completed response metadata for session affinity" + "解析已完成响应 metadata 用于 session affinity 失败" ); return; } @@ -727,14 +727,15 @@ async fn apply_upstream_account_retry_with_deps( } => { let cooldown_until = Utc::now() + Duration::seconds(retry_after_seconds as i64); if let Some(repo) = deps.account_repository.as_ref() { - if repo + if let Err(error) = repo .set_quota_cooldown_until(&account.id, cooldown_until) .await - .is_err() { tracing::warn!( + error = %error, account_id = %account.id, - "failed to persist quota cooldown" + cooldown_until = %cooldown_until, + "持久化 quota cooldown 失败" ); } } @@ -742,10 +743,11 @@ async fn apply_upstream_account_retry_with_deps( .lock() .await .mark_quota_limited_until(&account.id, cooldown_until); - if record_request_attempt(deps, &account.id).await.is_err() { + if let Err(error) = record_request_attempt(deps, &account.id).await { tracing::warn!( + error = ?error, account_id = %account.id, - "failed to record rate-limited account attempt" + "记录被 rate limit 的账户请求尝试失败" ); } } @@ -755,26 +757,24 @@ async fn apply_upstream_account_retry_with_deps( UpstreamAccountRetry::CloudflareChallenge { cooldown_seconds } => { let cooldown_until = Utc::now() + Duration::seconds(cooldown_seconds as i64); if let Some(cookie_repo) = deps.cookie_repository.as_ref() { - if cookie_repo - .delete_account_cookies(&account.id) - .await - .is_err() - { + if let Err(error) = cookie_repo.delete_account_cookies(&account.id).await { tracing::warn!( + error = %error, account_id = %account.id, - "failed to clear Cloudflare-blocked account cookies" + "清理 Cloudflare 阻断账户 cookies 失败" ); } } if let Some(repo) = deps.account_repository.as_ref() { - if repo + if let Err(error) = repo .set_cloudflare_cooldown_until(&account.id, cooldown_until) .await - .is_err() { tracing::warn!( + error = %error, account_id = %account.id, - "failed to persist Cloudflare cooldown" + cooldown_until = %cooldown_until, + "持久化 Cloudflare cooldown 失败" ); } } @@ -795,10 +795,12 @@ async fn set_account_status( status: AccountStatus, ) { if let Some(repo) = deps.account_repository.as_ref() { - if repo.set_status(&account.id, status).await.is_err() { + if let Err(error) = repo.set_status(&account.id, status).await { tracing::warn!( + error = %error, account_id = %account.id, - "failed to persist upstream account status" + status = ?status, + "持久化上游账户状态失败" ); } } @@ -843,7 +845,7 @@ async fn responses_websocket_stream( &log_context, retry.status(), EventLevel::Warn, - "v1 responses websocket history request kept on original account", + "v1 responses WebSocket history 请求保持原账户", websocket_history_retry_metadata(retry, true), ) .await; @@ -861,7 +863,7 @@ async fn responses_websocket_stream( &log_context, retry.status(), EventLevel::Warn, - "v1 responses websocket upstream retrying with fallback account", + "v1 responses WebSocket 上游请求将使用备用账户重试", retry.metadata(true), ) .await; @@ -878,7 +880,7 @@ async fn responses_websocket_stream( &log_context, error_response.0, EventLevel::Error, - "v1 responses websocket stream upstream request failed", + "v1 responses WebSocket stream 上游请求失败", json!({"stream": true, "transport": "websocket"}), ) .await; @@ -897,7 +899,7 @@ async fn responses_websocket_stream( &log_context, StatusCode::INTERNAL_SERVER_ERROR, EventLevel::Error, - "v1 responses websocket stream cookie persistence failed", + "v1 responses WebSocket stream 持久化 cookie 失败", json!({"stream": true, "transport": "websocket", "cookieStoreError": true}), ) .await; @@ -1059,7 +1061,7 @@ impl StreamAudit { let body = String::from_utf8_lossy(body); let mut status = StatusCode::OK; let mut level = EventLevel::Info; - let mut message = "v1 responses stream completed"; + let mut message = "v1 responses stream 已完成"; let mut metadata = match extract_sse_usage(&body) { Ok(usage) => { if let Some(usage) = usage { @@ -1068,7 +1070,7 @@ impl StreamAudit { .is_err() { level = EventLevel::Warn; - message = "v1 responses stream completed with usage store error"; + message = "v1 responses stream 已完成但 usage 存储失败"; json!({"stream": true, "usage": usage, "usageStoreError": true}) } else { json!({"stream": true, "usage": usage}) @@ -1079,7 +1081,7 @@ impl StreamAudit { } Err(error) => { level = EventLevel::Warn; - message = "v1 responses stream completed with invalid SSE usage"; + message = "v1 responses stream 已完成但 SSE usage 无效"; json!({"stream": true, "sseParseError": error.to_string()}) } }; @@ -1088,13 +1090,13 @@ impl StreamAudit { // SSE 响应头已发出，HTTP 状态不能回滚；用终止事件透传给客户端，并在审计里标记上游失败。 status = StatusCode::BAD_GATEWAY; level = EventLevel::Error; - message = "v1 responses stream upstream SSE failed"; + message = "v1 responses stream 上游 SSE 失败"; failure.extend_metadata(&mut metadata); } Ok(None) => {} Err(error) => { level = EventLevel::Warn; - message = "v1 responses stream completed with invalid SSE failure metadata"; + message = "v1 responses stream 已完成但 SSE 失败 metadata 无效"; metadata = json!({"stream": true, "sseParseError": error.to_string()}); } } @@ -1117,7 +1119,7 @@ impl StreamAudit { &self.context, StatusCode::BAD_GATEWAY, EventLevel::Error, - "v1 responses stream transport failed", + "v1 responses stream transport 失败", json!({"stream": true, "transportError": error.to_string()}), ) .await; @@ -1173,7 +1175,7 @@ impl WebSocketStreamAudit { let mut status = StatusCode::OK; let mut level = EventLevel::Info; - let mut message = "v1 responses websocket stream completed"; + let mut message = "v1 responses WebSocket stream 已完成"; let mut metadata = match usage_result { Ok(Some(usage)) => { if record_usage_with_deps(&self.deps, &self.context.account_id, usage) @@ -1181,7 +1183,7 @@ impl WebSocketStreamAudit { .is_err() { level = EventLevel::Warn; - message = "v1 responses websocket stream completed with usage store error"; + message = "v1 responses WebSocket stream 已完成但 usage 存储失败"; json!({ "stream": true, "transport": "websocket", @@ -1206,7 +1208,7 @@ impl WebSocketStreamAudit { }), Err(error) => { level = EventLevel::Warn; - message = "v1 responses websocket stream completed with invalid SSE usage"; + message = "v1 responses WebSocket stream 已完成但 SSE usage 无效"; json!({ "stream": true, "transport": "websocket", @@ -1220,14 +1222,13 @@ impl WebSocketStreamAudit { // SSE 响应头已经发给客户端，HTTP 状态不能回滚，只能在审计中标记上游失败。 status = StatusCode::BAD_GATEWAY; level = EventLevel::Error; - message = "v1 responses websocket stream upstream SSE failed"; + message = "v1 responses WebSocket stream 上游 SSE 失败"; failure.extend_metadata(&mut metadata); } Ok(None) => {} Err(error) => { level = EventLevel::Warn; - message = - "v1 responses websocket stream completed with invalid SSE failure metadata"; + message = "v1 responses WebSocket stream 已完成但 SSE 失败 metadata 无效"; metadata = json!({ "stream": true, "transport": "websocket", @@ -1255,7 +1256,7 @@ impl WebSocketStreamAudit { &self.context, StatusCode::BAD_GATEWAY, EventLevel::Error, - "v1 responses websocket stream transport failed", + "v1 responses WebSocket stream transport 失败", json!({ "stream": true, "transport": "websocket", @@ -1329,6 +1330,6 @@ async fn log_codex_upstream_response_with_deps( event.latency_ms = Some(context.latency_ms()); event.metadata = metadata; if let Err(error) = repo.insert(event).await { - tracing::warn!(?error, "failed to insert v1 response event log"); + tracing::warn!(error = %error, "写入 v1 response 事件日志失败"); } } diff --git a/src/codex/serving/dispatch/refresh.rs b/src/codex/serving/dispatch/refresh.rs index f927050ac..f3bbc3b87 100644 --- a/src/codex/serving/dispatch/refresh.rs +++ b/src/codex/serving/dispatch/refresh.rs @@ -108,7 +108,7 @@ async fn log_account_refresh_failure( let mut event = EventLog::new( "account.refresh", EventLevel::Warn, - "account refresh failed after upstream 401", + "上游 401 后账户刷新失败", ); event.request_id = Some(request_id.to_string()); event.account_id = Some(account.id.clone()); @@ -121,7 +121,7 @@ async fn log_account_refresh_failure( "accountStatus": status.map(account_status_value), }); if let Err(error) = repo.insert(event).await { - tracing::warn!(?error, "failed to insert account refresh event log"); + tracing::warn!(error = %error, "写入账户刷新事件日志失败"); } } diff --git a/src/codex/serving/responses.rs b/src/codex/serving/responses.rs index 91569b283..9692118fe 100644 --- a/src/codex/serving/responses.rs +++ b/src/codex/serving/responses.rs @@ -175,7 +175,7 @@ impl ResponsesService { &log_context, StatusCode::OK, EventLevel::Warn, - "v1 responses empty response, retrying", + "v1 responses 空响应，准备重试", json!({ "stream": false, "emptyResponse": true, @@ -190,7 +190,7 @@ impl ResponsesService { &log_context, StatusCode::BAD_GATEWAY, EventLevel::Error, - "v1 responses empty response after max retries", + "v1 responses 空响应重试次数已耗尽", json!({ "stream": false, "emptyResponse": true, @@ -217,13 +217,13 @@ impl ResponsesService { self.upstream.apply_account_retry(&account, retry).await; self.upstream .log_response( - &log_context, - retry.status(), - EventLevel::Warn, - "v1 responses websocket history request kept on original account", - websocket_history_retry_metadata(retry, false), - ) - .await; + &log_context, + retry.status(), + EventLevel::Warn, + "v1 responses WebSocket history 请求保持原账户", + websocket_history_retry_metadata(retry, false), + ) + .await; } else { let fallback = self .upstream @@ -239,7 +239,7 @@ impl ResponsesService { &log_context, retry.status(), EventLevel::Warn, - "v1 responses upstream retrying with fallback account", + "v1 responses 上游请求将使用备用账户重试", retry.metadata(false), ) .await; @@ -256,7 +256,7 @@ impl ResponsesService { &log_context, error_response.0, EventLevel::Error, - "v1 responses upstream request failed", + "v1 responses 上游请求失败", json!({"stream": false}), ) .await; @@ -275,7 +275,7 @@ impl ResponsesService { &log_context, StatusCode::INTERNAL_SERVER_ERROR, EventLevel::Error, - "v1 responses cookie persistence failed", + "v1 responses 持久化 cookie 失败", json!({"stream": false, "cookieStoreError": true}), ) .await; @@ -300,7 +300,7 @@ impl ResponsesService { &log_context, StatusCode::INTERNAL_SERVER_ERROR, EventLevel::Error, - "v1 responses usage persistence failed", + "v1 responses 持久化 usage 失败", json!({"stream": false, "usage": usage, "usageStoreError": true}), ) .await; @@ -331,7 +331,7 @@ impl ResponsesService { &log_context, StatusCode::OK, EventLevel::Info, - "v1 responses completed", + "v1 responses 已完成", json!({"stream": false, "usage": response.usage}), ) .await; @@ -343,7 +343,7 @@ impl ResponsesService { &log_context, StatusCode::BAD_GATEWAY, EventLevel::Error, - "v1 responses empty response", + "v1 responses 空响应", json!({"stream": false, "emptyResponse": true}), ) .await; @@ -362,7 +362,7 @@ impl ResponsesService { &log_context, StatusCode::BAD_GATEWAY, EventLevel::Error, - "v1 responses upstream SSE failed", + "v1 responses 上游 SSE 失败", failure.metadata(false), ) .await; @@ -381,7 +381,7 @@ impl ResponsesService { &log_context, StatusCode::BAD_GATEWAY, EventLevel::Warn, - "v1 responses completed event missing", + "v1 responses 缺少 completed 事件", json!({"stream": false, "usage": response.usage}), ) .await; @@ -400,7 +400,7 @@ impl ResponsesService { &log_context, StatusCode::BAD_GATEWAY, EventLevel::Warn, - "v1 responses invalid SSE response", + "v1 responses SSE 响应无效", json!({"stream": false, "sseParseError": error.to_string()}), ) .await; diff --git a/src/codex/tasks/model.rs b/src/codex/tasks/model.rs index 69463a061..8d9ca6933 100644 --- a/src/codex/tasks/model.rs +++ b/src/codex/tasks/model.rs @@ -43,7 +43,7 @@ impl ModelRefresher { let (shutdown_tx, mut shutdown_rx) = tokio::sync::mpsc::channel(1); tokio::spawn(async move { - info!("model refresher started"); + info!("模型刷新器已启动"); // 初始延迟后开始首次刷新 sleep(Duration::from_millis(self.initial_delay_ms)).await; @@ -55,7 +55,7 @@ impl ModelRefresher { while attempt < self.max_retries { tokio::select! { _ = shutdown_rx.recv() => { - info!("model refresher shutting down before first fetch"); + info!("模型刷新器在首次拉取前关闭"); return; } _ = async {} => {} @@ -67,7 +67,7 @@ impl ModelRefresher { info!( refreshed_plans = result.refreshed_plans, model_count = result.model_count, - "initial model refresh successful" + "首次模型刷新成功" ); has_fetched_once = true; break; @@ -82,9 +82,7 @@ impl ModelRefresher { ); if is_no_accounts { - warn!( - "initial model refresh skipped: no accounts available, will retry periodically" - ); + warn!("首次模型刷新跳过：没有可用账户，将进入周期重试"); break; } @@ -92,13 +90,13 @@ impl ModelRefresher { attempt, max_retries = self.max_retries, error = ?e, - "initial model refresh failed" + "首次模型刷新失败" ); if attempt < self.max_retries { tokio::select! { _ = shutdown_rx.recv() => { - info!("model refresher shutting down during retry wait"); + info!("模型刷新器在重试等待期间关闭"); return; } _ = sleep(Duration::from_millis(self.retry_delay_ms)) => {} @@ -109,9 +107,7 @@ impl ModelRefresher { } if !has_fetched_once { - warn!( - "model refresher failed all initial attempts, falling back to periodic refresh" - ); + warn!("模型刷新器首次尝试全部失败，切换到周期刷新"); } // 定期刷新 @@ -124,7 +120,7 @@ impl ModelRefresher { self.tick().await; } _ = shutdown_rx.recv() => { - info!("model refresher shutting down"); + info!("模型刷新器正在关闭"); break; } } @@ -142,11 +138,11 @@ impl ModelRefresher { refreshed_plans = result.refreshed_plans, model_count = result.model_count, failed_plans = result.failed_plans, - "models refreshed" + "模型列表已刷新" ); } Err(e) => { - warn!(error = ?e, "failed to refresh models"); + warn!(error = ?e, "刷新模型列表失败"); } } } diff --git a/src/codex/tasks/quota.rs b/src/codex/tasks/quota.rs index 2f261b3e5..c1c5f7d08 100644 --- a/src/codex/tasks/quota.rs +++ b/src/codex/tasks/quota.rs @@ -39,10 +39,7 @@ impl QuotaRefresher { let mut ticker = interval(Duration::from_secs(self.interval_secs)); let mut last_refreshed: HashMap = HashMap::new(); - info!( - interval_secs = self.interval_secs, - "quota refresher started" - ); + info!(interval_secs = self.interval_secs, "quota 刷新器已启动"); loop { tokio::select! { @@ -50,7 +47,7 @@ impl QuotaRefresher { self.tick(&mut last_refreshed).await; } _ = shutdown_rx.recv() => { - info!("quota refresher shutting down"); + info!("quota 刷新器正在关闭"); break; } } @@ -68,13 +65,13 @@ impl QuotaRefresher { let accounts_to_refresh = self.account_service.list_quota_locked_accounts().await; if accounts_to_refresh.is_empty() { - debug!("no quota-locked accounts found"); + debug!("未找到 quota 锁定账户"); return; } info!( count = accounts_to_refresh.len(), - "found quota-locked accounts to refresh" + "发现需要刷新 quota 的锁定账户" ); for account_id in accounts_to_refresh { @@ -83,13 +80,13 @@ impl QuotaRefresher { if now.duration_since(last_time) < min_interval { debug!( account_id = %account_id, - "skipping - recently refreshed" + "账户最近已刷新，跳过 quota 刷新" ); continue; } } - info!(account_id = %account_id, "actively refreshing quota"); + info!(account_id = %account_id, "正在主动刷新 quota"); // 记录刷新时间 last_refreshed.insert(account_id.clone(), now); @@ -102,13 +99,13 @@ impl QuotaRefresher { .await { Ok(_) => { - info!(account_id = %account_id, "quota refreshed successfully"); + info!(account_id = %account_id, "quota 刷新成功"); } Err(e) => { warn!( account_id = %account_id, error = ?e, - "failed to refresh quota" + "quota 刷新失败" ); } } diff --git a/src/codex/tasks/refresh.rs b/src/codex/tasks/refresh.rs index c4be90260..bf9217b50 100644 --- a/src/codex/tasks/refresh.rs +++ b/src/codex/tasks/refresh.rs @@ -61,7 +61,7 @@ impl RefreshScheduler { /// 为所有账户调度刷新 pub async fn schedule_all(&self) -> SchedulerResult<()> { if !self.config.auth.refresh_enabled { - info!("Auto-refresh disabled (refresh_enabled = false)"); + info!("自动刷新已关闭（refresh_enabled = false）"); return Ok(()); } @@ -89,7 +89,7 @@ impl RefreshScheduler { match account.status { AccountStatus::Refreshing => { // 崩溃恢复：进程终止时正在刷新 - info!(account_id = %account.id, "Recovering from 'refreshing' state"); + info!(account_id = %account.id, "正在从 refreshing 状态恢复"); self.do_refresh(account.id.clone()).await; } AccountStatus::Expired => { @@ -99,7 +99,7 @@ impl RefreshScheduler { info!( account_id = %account.id, delay_secs = delay.as_secs(), - "Expired account, scheduling recovery attempt" + "账户已过期，已调度恢复尝试" ); self.schedule_recovery(account.id.clone(), delay).await; } @@ -119,7 +119,7 @@ impl RefreshScheduler { pub async fn trigger_refresh_now(&self, account_id: String) -> SchedulerResult<()> { // 检查账户是否正在刷新 if self.is_refreshing(&account_id).await { - debug!(account_id = %account_id, "Account already refreshing, skipping"); + debug!(account_id = %account_id, "账户正在刷新，跳过本次触发"); return Ok(()); } @@ -140,7 +140,7 @@ impl RefreshScheduler { let exp = match self.parse_jwt_exp(&access_token) { Some(exp) => exp, None => { - warn!(account_id = %account_id, "Failed to parse JWT exp, skipping schedule"); + warn!(account_id = %account_id, "解析 JWT exp 失败，跳过刷新调度"); return Ok(()); } }; @@ -151,7 +151,7 @@ impl RefreshScheduler { if refresh_at <= now { // 已经过了刷新时间 - 立即尝试刷新 - debug!(account_id = %account_id, "Token already past refresh time, refreshing immediately"); + debug!(account_id = %account_id, "token 已超过刷新时间，立即刷新"); self.do_refresh(account_id).await; return Ok(()); } @@ -160,7 +160,7 @@ impl RefreshScheduler { info!( account_id = %account_id, delay_secs = delay.as_secs(), - "Refresh scheduled" + "已调度 token 刷新" ); // 启动定时器 - 直接刷新，不再调用 do_refresh @@ -204,7 +204,7 @@ impl RefreshScheduler { let exp = match self.parse_jwt_exp(&access_token) { Some(exp) => exp, None => { - warn!(account_id = %account_id, "Failed to parse JWT exp, skipping schedule"); + warn!(account_id = %account_id, "解析 JWT exp 失败，跳过刷新调度"); return Ok(()); } }; @@ -215,7 +215,7 @@ impl RefreshScheduler { if refresh_at <= now { // 已经过了刷新时间 - debug!(account_id = %account_id, "Token already past refresh time after refresh"); + debug!(account_id = %account_id, "刷新后 token 仍已超过下次刷新时间"); return Ok(()); } @@ -223,7 +223,7 @@ impl RefreshScheduler { info!( account_id = %account_id, delay_secs = delay.as_secs(), - "Next refresh scheduled" + "已调度下一次 token 刷新" ); // 启动定时器 @@ -280,7 +280,7 @@ impl RefreshScheduler { { let mut in_flight = self.in_flight.write().await; if in_flight.contains_key(&account_id) { - debug!(account_id = %account_id, "Already in flight, skipping"); + debug!(account_id = %account_id, "刷新任务已在执行，跳过"); return; } in_flight.insert(account_id.clone(), Instant::now()); @@ -293,20 +293,21 @@ impl RefreshScheduler { self.in_flight.write().await.remove(&account_id); if let Err(e) = result { - error!(account_id = %account_id, error = %e, "Refresh failed"); + error!(account_id = %account_id, error = %e, "token 刷新失败"); } } /// 内部刷新逻辑（带重试和错误处理） + #[tracing::instrument(skip(self), fields(account_id = %account_id))] async fn do_refresh_inner(&self, account_id: &str) -> SchedulerResult<()> { - info!(account_id = %account_id, "Starting token refresh"); + info!(account_id = %account_id, "开始刷新 token"); let mut permanent_hits = 0; for attempt in 1..=MAX_ATTEMPTS { match self.account_service.refresh_account(account_id).await { Ok(_result) => { - info!(account_id = %account_id, "Token refreshed successfully"); + info!(account_id = %account_id, "token 刷新成功"); // 获取新令牌并重新调度下次刷新 if let Ok(accounts) = self .account_service @@ -339,7 +340,7 @@ impl RefreshScheduler { error!( account_id = %account_id, hits = permanent_hits, - "Permanent failure detected" + "检测到永久刷新失败" ); // refresh_account 已经标记了状态，这里不需要再次标记 return Err(SchedulerError::AccountNotFound(format!( @@ -351,7 +352,7 @@ impl RefreshScheduler { account_id = %account_id, hits = permanent_hits, threshold = PERMANENT_THRESHOLD, - "Permanent error detected, retrying" + "检测到永久错误，继续重试确认" ); } @@ -367,7 +368,7 @@ impl RefreshScheduler { max_attempts = MAX_ATTEMPTS, delay_secs = delay.as_secs(), error = %e, - "Refresh attempt failed, retrying" + "token 刷新尝试失败，准备重试" ); tokio::time::sleep(delay).await; } else { @@ -375,7 +376,7 @@ impl RefreshScheduler { account_id = %account_id, attempts = MAX_ATTEMPTS, error = %e, - "All refresh attempts exhausted" + "token 刷新重试次数已耗尽" ); // 调度恢复尝试（不在循环内调用 await，避免递归） let account_id_recovery = account_id.to_string(); @@ -433,7 +434,7 @@ impl RefreshScheduler { } timers.clear(); - info!("Refresh scheduler shut down gracefully"); + info!("token 刷新调度器已关闭"); }); SchedulerHandle::new(shutdown_tx) diff --git a/src/config/types.rs b/src/config/types.rs index 7f2e5c291..9367bd3fb 100644 --- a/src/config/types.rs +++ b/src/config/types.rs @@ -109,10 +109,10 @@ fn default_admin_password() -> String { } #[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] pub struct LoggingConfig { pub directory: String, - pub max_file_bytes: u64, - pub retention_days: u64, + pub retention_days: usize, pub enabled: bool, pub capacity: u32, pub capture_body: bool, diff --git a/src/main.rs b/src/main.rs index ed9b431b3..3915e1c4c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -9,32 +9,30 @@ use tokio::signal; async fn main() -> anyhow::Result<()> { let config = AppConfig::load()?; - let _log_writer = init_tracing(RotationConfig::new( + let _log_guard = init_tracing(RotationConfig::new( &config.logging.directory, - config.logging.max_file_bytes, config.logging.retention_days, ))?; let host = config.server.host.clone(); let port = config.server.port; let (state, db_pool, restored_accounts) = build_state(config.clone()).await?; - tracing::info!(restored_accounts, "account pool restored from sqlite"); + tracing::info!(account_count = restored_accounts, "账户池已从 SQLite 恢复"); let background_tasks = start_background_tasks(&state, db_pool.clone(), &config).await; let app = build_router(state); let listener = tokio::net::TcpListener::bind((host.as_str(), port)).await?; - tracing::info!(host, port, "codex-proxy-rs listening"); + tracing::info!(host, port, "codex-proxy-rs 已开始监听"); - // 运行服务器，同时监听关闭信号 tokio::select! { result = axum::serve(listener, app) => { if let Err(e) = result { - tracing::error!(error = %e, "server error"); + tracing::error!(error = %e, "服务器运行失败"); } } _ = signal::ctrl_c() => { - tracing::info!("received shutdown signal"); + tracing::info!("收到关闭信号"); } } diff --git a/src/runtime/bootstrap.rs b/src/runtime/bootstrap.rs index 9d170a892..1678afda0 100644 --- a/src/runtime/bootstrap.rs +++ b/src/runtime/bootstrap.rs @@ -50,7 +50,7 @@ pub async fn build_state(config: AppConfig) -> BootstrapResult<(AppState, Sqlite version = %fp.app_version, build = %fp.build_number, source = "database", - "loaded fingerprint for requests" + "已从数据库加载请求 fingerprint" ); fp } @@ -60,14 +60,14 @@ pub async fn build_state(config: AppConfig) -> BootstrapResult<(AppState, Sqlite version = %fp.app_version, build = %fp.build_number, source = "default", - "using default fingerprint for requests" + "使用默认请求 fingerprint" ); fp } Err(e) => { tracing::warn!( error = %e, - "failed to load fingerprint from database, using default" + "从数据库加载 fingerprint 失败，使用默认值" ); Fingerprint::default_codex_desktop() } @@ -117,7 +117,7 @@ async fn ensure_default_admin_exists( tracing::info!( username = %config.admin.default_username, - "created default admin user" + "已创建默认管理员用户" ); } diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs index 2b91ab657..d6641e904 100644 --- a/src/runtime/mod.rs +++ b/src/runtime/mod.rs @@ -3,4 +3,4 @@ pub mod router; pub mod state; pub mod tasks; -pub use router::build_router; +pub use router::{build_router, http_trace_layer}; diff --git a/src/runtime/router.rs b/src/runtime/router.rs index 6b2c0e1f2..889308f77 100644 --- a/src/runtime/router.rs +++ b/src/runtime/router.rs @@ -1,5 +1,16 @@ -use axum::{middleware::from_fn, routing::get, Router}; -use tower_http::trace::TraceLayer; +use std::time::Duration; + +use axum::{ + http::{Request, Response}, + middleware::from_fn, + routing::get, + Router, +}; +use tower_http::{ + classify::ServerErrorsFailureClass, + trace::{HttpMakeClassifier, MakeSpan, OnFailure, OnRequest, OnResponse, TraceLayer}, +}; +use tracing::Span; use crate::{ admin::http as admin_http, @@ -7,11 +18,36 @@ use crate::{ diagnostics::{debug_fingerprint, debug_upstream, diagnostics}, router as serving_http, }, - platform::http::{health::health, middleware::attach_request_id}, + platform::http::{ + health::health, + middleware::{attach_request_id, RequestId}, + }, }; use super::state::AppState; +pub type HttpTraceLayer = TraceLayer< + HttpMakeClassifier, + HttpMakeSpan, + HttpOnRequest, + HttpOnResponse, + (), + (), + HttpOnFailure, +>; + +#[derive(Debug, Clone, Copy)] +pub struct HttpMakeSpan; + +#[derive(Debug, Clone, Copy)] +pub struct HttpOnRequest; + +#[derive(Debug, Clone, Copy)] +pub struct HttpOnResponse; + +#[derive(Debug, Clone, Copy)] +pub struct HttpOnFailure; + pub fn build_router(state: AppState) -> Router { Router::new() .route("/health", get(health)) @@ -21,6 +57,64 @@ pub fn build_router(state: AppState) -> Router { .merge(serving_http::router()) .merge(admin_http::router()) .with_state(state) - .layer(TraceLayer::new_for_http()) + .layer(http_trace_layer()) .layer(from_fn(attach_request_id)) } + +pub fn http_trace_layer() -> HttpTraceLayer { + TraceLayer::new_for_http() + .make_span_with(HttpMakeSpan) + .on_request(HttpOnRequest) + .on_response(HttpOnResponse) + .on_body_chunk(()) + .on_eos(()) + .on_failure(HttpOnFailure) +} + +impl MakeSpan for HttpMakeSpan { + fn make_span(&mut self, request: &Request) -> Span { + let request_id = request + .extensions() + .get::() + .map(RequestId::as_str) + .unwrap_or("missing"); + + tracing::info_span!( + "http_request", + request_id = %request_id, + method = %request.method(), + uri = %request.uri(), + status = tracing::field::Empty, + latency_ms = tracing::field::Empty, + ) + } +} + +impl OnRequest for HttpOnRequest { + fn on_request(&mut self, _request: &Request, _span: &Span) { + tracing::info!("收到 HTTP 请求"); + } +} + +impl OnResponse for HttpOnResponse { + fn on_response(self, response: &Response, latency: Duration, span: &Span) { + let status = response.status().as_u16(); + let latency_ms = latency.as_millis() as u64; + span.record("status", status); + span.record("latency_ms", latency_ms); + tracing::info!(status, latency_ms, "HTTP 请求完成"); + } +} + +impl OnFailure for HttpOnFailure { + fn on_failure( + &mut self, + failure_class: ServerErrorsFailureClass, + latency: Duration, + span: &Span, + ) { + let latency_ms = latency.as_millis() as u64; + span.record("latency_ms", latency_ms); + tracing::warn!(error = %failure_class, latency_ms, "HTTP 请求失败"); + } +} diff --git a/src/runtime/tasks/coordinator.rs b/src/runtime/tasks/coordinator.rs index d1ec53827..205621e2c 100644 --- a/src/runtime/tasks/coordinator.rs +++ b/src/runtime/tasks/coordinator.rs @@ -25,12 +25,12 @@ impl BackgroundTaskCoordinator { } pub async fn shutdown(self) { - tracing::info!("shutting down background tasks"); + tracing::info!("正在关闭后台任务"); for (name, handle) in self.handles { handle.shutdown().await; - tracing::info!(task = name, "background task stopped"); + tracing::info!(task = name, "后台任务已停止"); } - tracing::info!("all background tasks stopped"); + tracing::info!("所有后台任务已停止"); } } @@ -48,7 +48,7 @@ pub async fn start_background_tasks( tracing::info!( version = %fp.app_version, build = %fp.build_number, - "loaded fingerprint from database (auto_update)" + "已从数据库加载 auto_update fingerprint" ); fp } @@ -57,12 +57,12 @@ pub async fn start_background_tasks( tracing::info!( version = %fp.app_version, build = %fp.build_number, - "using default fingerprint" + "使用默认 fingerprint" ); fp } Err(e) => { - tracing::warn!(error = %e, "failed to load fingerprint from database, using default"); + tracing::warn!(error = %e, "从数据库加载 fingerprint 失败，使用默认值"); Fingerprint::default_codex_desktop() } }; @@ -78,24 +78,24 @@ pub async fn start_background_tasks( "fingerprint_update", SchedulerHandle::from_join_handle(update_handle), ); - tracing::info!("fingerprint update checker started"); + tracing::info!("fingerprint 更新检查器已启动"); let refresh_scheduler = RefreshScheduler::new(state.account_service(), config.clone()); coordinator.push("refresh", refresh_scheduler.start().await); - tracing::info!("refresh scheduler started"); + tracing::info!("token 刷新调度器已启动"); let session_cleanup = SessionCleanupScheduler::new(db_pool, config.admin.session_cleanup_interval_secs); coordinator.push("session_cleanup", session_cleanup.start()); - tracing::info!("session cleanup scheduler started"); + tracing::info!("会话清理调度器已启动"); let quota_refresher = QuotaRefresher::new(state.account_service()); coordinator.push("quota", quota_refresher.start()); - tracing::info!("quota refresher started"); + tracing::info!("quota 刷新器已启动"); let model_refresher = ModelRefresher::new(state.model_service()); coordinator.push("model", model_refresher.start()); - tracing::info!("model refresher started"); + tracing::info!("模型刷新器已启动"); coordinator } diff --git a/tests/admin/api_keys_route.rs b/tests/admin/api_keys_route.rs index 43f46922a..c84e07dec 100644 --- a/tests/admin/api_keys_route.rs +++ b/tests/admin/api_keys_route.rs @@ -73,7 +73,6 @@ fn test_config(database_url: String) -> AppConfig { }, logging: LoggingConfig { directory: "logs".to_string(), - max_file_bytes: 10_485_760, retention_days: 14, enabled: false, capacity: 2_000, diff --git a/tests/admin/login_route.rs b/tests/admin/login_route.rs index 03e03aa5c..0e4099f45 100644 --- a/tests/admin/login_route.rs +++ b/tests/admin/login_route.rs @@ -71,7 +71,6 @@ fn test_config(database_url: String) -> AppConfig { }, logging: LoggingConfig { directory: "logs".to_string(), - max_file_bytes: 10_485_760, retention_days: 14, enabled: false, capacity: 2_000, diff --git a/tests/admin/logs_route.rs b/tests/admin/logs_route.rs index 8151cfa2a..64eb68990 100644 --- a/tests/admin/logs_route.rs +++ b/tests/admin/logs_route.rs @@ -74,7 +74,6 @@ fn test_config(database_url: String) -> AppConfig { }, logging: LoggingConfig { directory: "logs".to_string(), - max_file_bytes: 10_485_760, retention_days: 14, enabled: false, capacity: 2_000, diff --git a/tests/admin/models_route.rs b/tests/admin/models_route.rs index c844b058a..da5cd582a 100644 --- a/tests/admin/models_route.rs +++ b/tests/admin/models_route.rs @@ -81,7 +81,6 @@ fn test_config(database_url: String, base_url: String) -> AppConfig { }, logging: LoggingConfig { directory: "logs".to_string(), - max_file_bytes: 10_485_760, retention_days: 14, enabled: false, capacity: 2_000, diff --git a/tests/admin/settings_route.rs b/tests/admin/settings_route.rs index 287106ba7..27780579d 100644 --- a/tests/admin/settings_route.rs +++ b/tests/admin/settings_route.rs @@ -73,7 +73,6 @@ fn test_config(database_url: String) -> AppConfig { }, logging: LoggingConfig { directory: "logs".to_string(), - max_file_bytes: 10_485_760, retention_days: 14, enabled: true, capacity: 2_000, diff --git a/tests/admin/usage_stats_route.rs b/tests/admin/usage_stats_route.rs index 557dc4078..b6febbbdd 100644 --- a/tests/admin/usage_stats_route.rs +++ b/tests/admin/usage_stats_route.rs @@ -71,7 +71,6 @@ fn test_config(database_url: String) -> AppConfig { }, logging: LoggingConfig { directory: "logs".to_string(), - max_file_bytes: 10_485_760, retention_days: 14, enabled: false, capacity: 2_000, diff --git a/tests/codex_operations/log_rotation.rs b/tests/codex_operations/log_rotation.rs index cf486888b..66cd7d280 100644 --- a/tests/codex_operations/log_rotation.rs +++ b/tests/codex_operations/log_rotation.rs @@ -1,34 +1,32 @@ use std::{fs, io::Write}; -use tracing_subscriber::fmt::MakeWriter; - -use codex_proxy_rs::codex::logs::rotation::{RotatingLogWriter, RotationConfig}; +use codex_proxy_rs::codex::logs::rotation::{build_file_appender, RotationConfig}; #[test] -fn rotating_writer_splits_files_when_size_limit_is_reached() { +fn rolling_appender_should_write_daily_codex_log_file() { let dir = tempfile::tempdir().unwrap(); - let writer = RotatingLogWriter::new(RotationConfig::new(dir.path(), 16, 14)).unwrap(); + let config = RotationConfig::new(dir.path(), 14); + let mut appender = build_file_appender(&config).unwrap(); - { - let mut file = writer.make_writer(); - file.write_all(b"first log line\n").unwrap(); - file.write_all(b"second log line\n").unwrap(); - file.flush().unwrap(); - } + appender.write_all(b"hello\n").unwrap(); + appender.flush().unwrap(); - let log_files = fs::read_dir(dir.path()) + let names = fs::read_dir(dir.path()) .unwrap() - .filter_map(Result::ok) - .filter(|entry| { - entry - .file_name() - .to_str() - .is_some_and(|name| name.starts_with("codex-proxy-rs") && name.ends_with(".log")) - }) - .count(); + .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned()) + .collect::>(); assert!( - log_files >= 2, - "expected rotated log files, found {log_files}" + names + .iter() + .any(|name| name.starts_with("codex-proxy-rs.") && name.ends_with(".log")), + "expected a daily codex log file, found {names:?}" ); } + +#[test] +fn rotation_config_should_not_accept_size_limit() { + let config = RotationConfig::new("logs", 14); + + assert_eq!(config.retention_days, 14); +} diff --git a/tests/codex_serving/chat_completions.rs b/tests/codex_serving/chat_completions.rs index 5793a46e2..05282e89a 100644 --- a/tests/codex_serving/chat_completions.rs +++ b/tests/codex_serving/chat_completions.rs @@ -97,7 +97,6 @@ fn test_config(database_url: String, base_url: String) -> AppConfig { }, logging: LoggingConfig { directory: "logs".to_string(), - max_file_bytes: 10_485_760, retention_days: 14, enabled: false, capacity: 2_000, diff --git a/tests/codex_serving/routes_responses.rs b/tests/codex_serving/routes_responses.rs index ac15c0dd4..08a3ab709 100644 --- a/tests/codex_serving/routes_responses.rs +++ b/tests/codex_serving/routes_responses.rs @@ -80,7 +80,6 @@ fn test_config() -> AppConfig { }, logging: LoggingConfig { directory: "logs".to_string(), - max_file_bytes: 10_485_760, retention_days: 14, enabled: false, capacity: 2_000, diff --git a/tests/codex_serving/upstream_fallback.rs b/tests/codex_serving/upstream_fallback.rs index d8f08bfbd..a3cca0c74 100644 --- a/tests/codex_serving/upstream_fallback.rs +++ b/tests/codex_serving/upstream_fallback.rs @@ -639,7 +639,7 @@ async fn v1_responses_should_log_refresh_failure_after_401() { .unwrap(); let metadata: Value = serde_json::from_str(&refresh_event.2).unwrap(); assert_eq!(refresh_event.0, "warn"); - assert_eq!(refresh_event.1, "account refresh failed after upstream 401"); + assert_eq!(refresh_event.1, "上游 401 后账户刷新失败"); assert_eq!(metadata["failure"], "invalidGrant"); assert_eq!(metadata["accountStatus"], "expired"); assert_eq!(metadata["trigger"], "upstream_401"); diff --git a/tests/config.rs b/tests/config.rs index 34a5f43e2..a673bc833 100644 --- a/tests/config.rs +++ b/tests/config.rs @@ -1,4 +1,4 @@ -use codex_proxy_rs::config::AppConfig; +use codex_proxy_rs::config::{AppConfig, LoggingConfig}; use std::fs; #[test] @@ -60,7 +60,6 @@ admin: session_ttl_minutes: 1440 logging: directory: logs - max_file_bytes: 10485760 retention_days: 14 enabled: false capacity: 2000 @@ -87,5 +86,25 @@ logging: assert_eq!(cfg.model.default_model, "gpt-5.5"); assert_eq!(cfg.auth.max_concurrent_per_account, 3); assert_eq!(cfg.logging.directory, "local-logs"); - assert_eq!(cfg.logging.max_file_bytes, 10485760); + assert_eq!(cfg.logging.retention_days, 14); +} + +#[test] +fn logging_config_rejects_removed_size_rotation_field() { + let err = serde_yml::from_str::( + r#" +directory: logs +max_file_bytes: 10485760 +retention_days: 14 +enabled: false +capacity: 2000 +capture_body: false +"#, + ) + .unwrap_err(); + + assert!( + err.to_string().contains("max_file_bytes"), + "expected max_file_bytes to be rejected, got {err}" + ); } diff --git a/tests/runtime.rs b/tests/runtime.rs index 96e695a66..143724dfc 100644 --- a/tests/runtime.rs +++ b/tests/runtime.rs @@ -1,3 +1,4 @@ mod runtime { + mod http_trace; mod startup; } diff --git a/tests/runtime/http_trace.rs b/tests/runtime/http_trace.rs new file mode 100644 index 000000000..a5a19ecdb --- /dev/null +++ b/tests/runtime/http_trace.rs @@ -0,0 +1,93 @@ +use std::{ + io::{self, Write}, + sync::{Arc, Mutex}, +}; + +use axum::{ + body::Body, + http::{Request, StatusCode}, + middleware::from_fn, + routing::get, + Router, +}; +use tower::ServiceExt; +use tracing_subscriber::fmt::MakeWriter; + +use codex_proxy_rs::{platform::http::middleware::attach_request_id, runtime::http_trace_layer}; + +#[derive(Clone, Default)] +struct SharedLogBuffer { + inner: Arc>>, +} + +impl SharedLogBuffer { + fn content(&self) -> String { + let bytes = self.inner.lock().unwrap().clone(); + String::from_utf8(bytes).unwrap() + } +} + +impl<'a> MakeWriter<'a> for SharedLogBuffer { + type Writer = SharedLogWriter; + + fn make_writer(&'a self) -> Self::Writer { + SharedLogWriter { + inner: Arc::clone(&self.inner), + } + } +} + +struct SharedLogWriter { + inner: Arc>>, +} + +impl Write for SharedLogWriter { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.inner.lock().unwrap().extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +#[tokio::test] +async fn http_trace_should_include_request_id_and_completion_fields() { + 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 _guard = tracing::subscriber::set_default(subscriber); + let response = app + .oneshot( + Request::builder() + .uri("/trace-test") + .header("x-request-id", "req_trace") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::NO_CONTENT); + let output = logs.content(); + assert!( + output.contains("req_trace") + && output.contains("收到 HTTP 请求") + && output.contains("HTTP 请求完成") + && output.contains("\"status\":204") + && output.contains("latency_ms"), + "unexpected trace output: {output}" + ); +} diff --git a/tests/runtime/startup.rs b/tests/runtime/startup.rs index d9993e4d4..6f34cf6e7 100644 --- a/tests/runtime/startup.rs +++ b/tests/runtime/startup.rs @@ -67,7 +67,6 @@ fn test_config(database_url: String) -> AppConfig { }, logging: LoggingConfig { directory: "logs".to_string(), - max_file_bytes: 10_485_760, retention_days: 14, enabled: false, capacity: 2_000, diff --git a/tests/support/admin_accounts.rs b/tests/support/admin_accounts.rs index 9d730c309..67062a1d4 100644 --- a/tests/support/admin_accounts.rs +++ b/tests/support/admin_accounts.rs @@ -156,7 +156,6 @@ pub fn test_config(database_url: String) -> AppConfig { }, logging: LoggingConfig { directory: "logs".to_string(), - max_file_bytes: 10_485_760, retention_days: 14, enabled: false, capacity: 2_000, diff --git a/tests/support/upstream.rs b/tests/support/upstream.rs index bceca3673..0ab014864 100644 --- a/tests/support/upstream.rs +++ b/tests/support/upstream.rs @@ -116,7 +116,6 @@ pub fn test_config(database_url: String, base_url: String) -> AppConfig { }, logging: LoggingConfig { directory: "logs".to_string(), - max_file_bytes: 10_485_760, retention_days: 14, enabled: false, capacity: 2_000,