feat: harden logging and retention boundaries

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
zyycn
2026-07-11 04:07:34 +08:00
co-authored by Codex
parent 899454e6b5
commit 1652b60769
38 changed files with 1218 additions and 248 deletions
+23 -3
View File
@@ -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: '<set-a-long-random-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
```
构建并启动:
+13 -11
View File
@@ -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<Body>) -> 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<Body>, _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<Body>, _span: &Span) {
debug!("received HTTP request");
}
fn on_http_response(response: &Response<Body>, 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<Body>) -> Option<String> {
.map(|rid| rid.as_str().to_string())
}
fn client_ip(request: &Request<Body>) -> Option<&str> {
request.extensions().get::<ClientIp>().map(ClientIp::as_str)
}
fn sanitized_uri(request: &Request<Body>) -> String {
let uri = request.uri();
let path = uri.path();
+6 -3
View File
@@ -80,12 +80,15 @@ pub fn router() -> Router<AppState> {
/// 使用指定前端构建产物目录构造整个 HTTP 服务路由。
pub fn router_with_assets(dist_dir: impl AsRef<Path>) -> Router<AppState> {
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))
}
+53 -8
View File
@@ -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 {
+27 -8
View File
@@ -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<Option<LogGuard>, crate::infra::logging::LogError> {
if !config.logging.enabled {
return Ok(None);
}
fn init_logging(config: &AppConfig) -> Result<LogGuard, LogError> {
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() {
@@ -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(),
);
+38 -3
View File
@@ -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, "清理超限指纹更新历史失败"),
}
})
}
}
+3 -3
View File
@@ -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");
}
}
}
+14 -14
View File
@@ -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::<serde_json::Value>(&quota_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"
+1 -1
View File
@@ -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;
+476 -73
View File
@@ -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<Path>, retention_days: usize) -> Self {
pub fn new(
directory: impl AsRef<Path>,
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<RotationConfig>,
}
impl TracingConfig {
/// 创建 tracing 输出配置。
pub fn new(level: impl Into<String>, stdout: bool, file: Option<RotationConfig>) -> 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<WorkerGuard>,
}
/// 按中国自然日写入的日志 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<ChinaDailyFileAppender, LogError> {
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<LogGuard, LogError> {
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<LogGuard, LogError> {
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<S>(
writer: tracing_appender::non_blocking::NonBlocking,
) -> tracing_subscriber::fmt::Layer<
S,
tracing_subscriber::fmt::format::JsonFields,
tracing_subscriber::fmt::format::Format<tracing_subscriber::fmt::format::Json, ChinaLogTimer>,
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<Path>,
retention_days: usize,
now: DateTime<Utc>,
) -> Result<Self, LogError> {
let directory = directory.as_ref().to_path_buf();
fn open(config: RotationConfig, now: DateTime<Utc>) -> Result<Self, LogError> {
cleanup_managed_logs(&config, now)?;
let current_date = china_date(&now);
let writer = open_log_file(&directory, &current_date)?;
cleanup_old_logs(&directory, retention_days)?;
let (current_segment, bytes_written) = writable_segment(&config, &current_date)?;
let writer = open_log_file(&config.directory, &current_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, &current_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<usize> {
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<Utc>) -> String {
china_rfc3339_millis(&now)
}
fn open_log_file(directory: &Path, date: &str) -> io::Result<File> {
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<File> {
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<Vec<ManagedLogFile>> {
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::<Vec<_>>();
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<Utc>) -> 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<thread::JoinHandle<()>>,
}
impl DroppedLogMonitor {
fn start(mut counters: Vec<NamedErrorCounter>) -> io::Result<Self> {
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<mpsc::Receiver<()>>,
}
impl Write for FirstWriteBlocking {
fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
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))
);
}
}
+1 -1
View File
@@ -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)
@@ -1,4 +1,35 @@
//! 账号累计用量与时间桶查询。
use chrono::{DateTime, Utc};
pub mod query;
pub mod store;
/// 根据额度窗口边界漂移判断是否重置本地窗口计数。
pub fn should_reset_usage_window(
existing_reset_at: Option<DateTime<Utc>>,
existing_limit_window_seconds: Option<u64>,
new_reset_at: DateTime<Utc>,
new_limit_window_seconds: Option<u64>,
) -> 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
}
+30 -29
View File
@@ -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<u64> {
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<DateTime<Utc>>,
existing_limit_window_seconds: Option<u64>,
new_reset_at: DateTime<Utc>,
new_limit_window_seconds: Option<u64>,
) -> 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
}
+3 -3
View File
@@ -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");
}
}
@@ -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<u64, sqlx::Error> {
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<Fingerprint, sqlx::Error> {
@@ -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
}
+1 -1
View File
@@ -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,
+120
View File
@@ -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<Value> {
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()
+32 -12
View File
@@ -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::<LoggingConfig>(
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();
+7 -7
View File
@@ -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)
@@ -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()
@@ -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(
@@ -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()
@@ -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(
+10 -10
View File
@@ -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()
+89 -19
View File
@@ -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::<Vec<_>>();
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<String> {
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")
}
}
+13 -5
View File
@@ -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 },
}
}
@@ -0,0 +1 @@
mod store;
@@ -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));
}
+1
View File
@@ -1,3 +1,4 @@
pub(crate) mod account_usage;
pub(crate) mod ops;
pub(crate) mod rebuild;
pub(crate) mod usage;
@@ -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;
+8
View File
@@ -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。
构建并启动:
+15 -5
View File
@@ -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
+26 -2
View File
@@ -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:
+13 -4
View File
@@ -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 <path>`(§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 服务与健康检查依赖。
+4
View File
@@ -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`。
+26 -14
View File
@@ -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。