From 8f242e2875dcd7dbfed8a0d0f5f0970838210e03 Mon Sep 17 00:00:00 2001 From: Zyy Date: Fri, 4 Sep 2026 19:15:28 +0800 Subject: [PATCH] feat(logging): add opt-in request dumps Capture complete OpenAI Responses headers and payloads in a dedicated rotated log, keep the target out of normal sinks, and default the sensitive switch to off. Co-authored-by: Codex --- .../gateway-api/src/openai/responses/http.rs | 13 +++ .../src/openai/responses/websocket/mod.rs | 40 +++++++- backend/crates/gateway-host/src/config.rs | 3 + backend/crates/gateway-host/src/logging.rs | 94 ++++++++++++++----- backend/crates/gateway-host/tests/config.rs | 1 + backend/crates/gateway-host/tests/logging.rs | 83 ++++++++++++++-- .../openai/src/transport/client_sse.rs | 48 +++++++++- deploy/config.example.yaml | 2 + 8 files changed, 244 insertions(+), 40 deletions(-) diff --git a/backend/crates/gateway-api/src/openai/responses/http.rs b/backend/crates/gateway-api/src/openai/responses/http.rs index 49db85519..735299eb2 100644 --- a/backend/crates/gateway-api/src/openai/responses/http.rs +++ b/backend/crates/gateway-api/src/openai/responses/http.rs @@ -83,6 +83,19 @@ pub(crate) async fn responses( Ok(started) => started, Err(error) => return gateway_error_response(&error), }; + tracing::info!( + target: "request_dump", + provider = "openai", + request_id = %started.request_id, + request_direction = "downstream", + request_transport = if streaming { "http_sse" } else { "http_json" }, + request_method = "POST", + request_path = "/v1/responses", + request_headers = ?headers, + request_body = %String::from_utf8_lossy(&body), + contains_sensitive_data = true, + "request dump (unredacted)" + ); let StartedExecution { stream, session, .. } = started; diff --git a/backend/crates/gateway-api/src/openai/responses/websocket/mod.rs b/backend/crates/gateway-api/src/openai/responses/websocket/mod.rs index 8759dd169..83ad91c2c 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket/mod.rs +++ b/backend/crates/gateway-api/src/openai/responses/websocket/mod.rs @@ -60,12 +60,14 @@ pub(crate) async fn responses_websocket( &headers, connect_info.map(|Extension(ConnectInfo(address))| address), ); + let request_headers = OpenAiRequestHeaders::from_headers(&headers); ResponsesWebSocketAdapter::new(service).upgrade_with_client_context( websocket, client, client_ip, user_agent, - OpenAiRequestHeaders::from_headers(&headers), + request_headers, + headers, ) } @@ -89,15 +91,29 @@ impl ResponsesWebSocketAdapter { client_ip: Option, user_agent: Option, request_headers: OpenAiRequestHeaders, + raw_headers: HeaderMap, ) -> Response { let connection_guard = match self.service.try_register_connection() { Ok(guard) => guard, Err(_) => return runtime_unavailable_response().into_response(), }; + let connection_id = self.service.next_request_id().replacen("req_", "ws_", 1); + tracing::info!( + target: "request_dump", + provider = "openai", + websocket_connection_id = %connection_id, + request_direction = "downstream", + request_transport = "websocket_handshake", + request_method = "GET", + request_path = "/v1/responses", + request_headers = ?raw_headers, + contains_sensitive_data = true, + "request dump (unredacted)" + ); let session = ResponsesWebSocketSession { service: self.service.clone(), client, - connection_id: self.service.next_request_id().replacen("req_", "ws_", 1), + connection_id, client_ip, user_agent, request_headers, @@ -171,6 +187,18 @@ async fn serve_responses_websocket(socket: WebSocket, session: ResponsesWebSocke } request_count = request_count.saturating_add(1); let correlation_id = Arc::::from(service.next_request_id()); + tracing::info!( + target: "request_dump", + provider = "openai", + websocket_connection_id = connection.id(), + correlation_id = %correlation_id, + request_index = request_count, + request_direction = "downstream", + request_transport = "websocket", + request_payload = %payload, + contains_sensitive_data = true, + "request dump (unredacted)" + ); let decoded = match decode_response_create_with_context(&payload, &request_headers) { Ok(decoded) => decoded.with_client_context(client_ip, user_agent.clone()), Err(error) => { @@ -219,6 +247,14 @@ async fn serve_responses_websocket(socket: WebSocket, session: ResponsesWebSocke continue; } }; + tracing::info!( + target: "request_dump", + provider = "openai", + websocket_connection_id = connection.id(), + correlation_id = %correlation_id, + request_id = %started.request_id, + "request dump correlation" + ); if forward_execution(&mut connection, started, &mut replay).await == ForwardOutcome::Disconnect diff --git a/backend/crates/gateway-host/src/config.rs b/backend/crates/gateway-host/src/config.rs index 05bc60120..5eef78b2f 100644 --- a/backend/crates/gateway-host/src/config.rs +++ b/backend/crates/gateway-host/src/config.rs @@ -118,6 +118,9 @@ pub struct LoggingConfig { pub level: String, pub stdout: bool, pub file: FileLoggingConfig, + /// 将完整请求原文写入独立诊断日志；默认关闭，因为内容包含凭据。 + #[serde(default)] + pub request_dump: bool, } impl LoggingConfig { diff --git a/backend/crates/gateway-host/src/logging.rs b/backend/crates/gateway-host/src/logging.rs index 44b755d28..c49525525 100644 --- a/backend/crates/gateway-host/src/logging.rs +++ b/backend/crates/gateway-host/src/logging.rs @@ -6,16 +6,19 @@ use std::io::{self, Write}; use std::path::{Path, PathBuf}; use chrono::Utc; -use tracing_appender::non_blocking::{NonBlockingBuilder, WorkerGuard}; +use tracing_appender::non_blocking::{NonBlocking, NonBlockingBuilder, WorkerGuard}; use tracing_subscriber::EnvFilter; use tracing_subscriber::{Layer as _, layer::SubscriberExt as _, util::SubscriberInitExt as _}; -use crate::config::LoggingConfig; +use crate::config::{FileLoggingConfig, LoggingConfig}; const LOG_FILE_PREFIX: &str = "codex-proxy-rs"; const OAUTH_RECOVERY_LOG_FILE_PREFIX: &str = "codex-proxy-rs-oath"; /// OAuth 恢复记录含原始 AT/RT，文件日志开启时必须落到独立文件。 const OAUTH_RECOVERY_LOG_TARGET: &str = "oauth_recovery"; +const REQUEST_DUMP_LOG_FILE_PREFIX: &str = "codex-proxy-rs-request-dump"; +/// 请求转储含完整凭据与正文，只允许在显式开启时写入独立文件。 +const REQUEST_DUMP_LOG_TARGET: &str = "request_dump"; /// non-blocking 日志 writer 的进程级守卫。 pub struct LogGuard { @@ -39,6 +42,7 @@ pub fn initialize_logging(config: &LoggingConfig) -> Result let directive = env::var("RUST_LOG").unwrap_or_else(|_| config.level.clone()); let file_filter = application_file_filter(&directive)?; let recovery_file_filter = oauth_recovery_file_filter(); + let request_dump_file_filter = request_dump_file_filter(); let stdout_filter = EnvFilter::try_new(stdout_filter_directive(&directive, config.file.enabled)) .map_err(|_| LogError::InvalidFilter)?; @@ -52,37 +56,30 @@ pub fn initialize_logging(config: &LoggingConfig) -> Result writer }); let (file_writer, recovery_file_writer) = if config.file.enabled { - let maximum_bytes = config - .file - .max_file_size_mb - .checked_mul(1024 * 1024) - .ok_or(LogError::SizeOverflow)?; - let file_writer = RotatingLogWriter::open( - config.file.directory.clone(), - LOG_FILE_PREFIX, - maximum_bytes, - config.file.retention_days, - config.file.max_files, - )?; - let recovery_file_writer = RotatingLogWriter::open( - config.file.directory.clone(), + let (file_writer, file_guard) = + rotating_file_writer(&config.file, LOG_FILE_PREFIX, "gateway-log-file")?; + let (recovery_file_writer, recovery_file_guard) = rotating_file_writer( + &config.file, OAUTH_RECOVERY_LOG_FILE_PREFIX, - maximum_bytes, - config.file.retention_days, - config.file.max_files, + "gateway-log-oauth-recovery-file", )?; - let (file_writer, file_guard) = NonBlockingBuilder::default() - .thread_name("gateway-log-file") - .finish(file_writer); - let (recovery_file_writer, recovery_file_guard) = NonBlockingBuilder::default() - .thread_name("gateway-log-oauth-recovery-file") - .finish(recovery_file_writer); guards.push(file_guard); guards.push(recovery_file_guard); (Some(file_writer), Some(recovery_file_writer)) } else { (None, None) }; + let request_dump_file_writer = if config.request_dump { + let (writer, guard) = rotating_file_writer( + &config.file, + REQUEST_DUMP_LOG_FILE_PREFIX, + "gateway-log-request-dump-file", + )?; + guards.push(guard); + Some(writer) + } else { + None + }; let stdout_layer = stdout_writer.map(|writer| { tracing_subscriber::fmt::layer() @@ -118,10 +115,24 @@ pub fn initialize_logging(config: &LoggingConfig) -> Result .with_span_list(true) .with_filter(recovery_file_filter) }); + let request_dump_file_layer = request_dump_file_writer.map(|writer| { + tracing_subscriber::fmt::layer() + .json() + .with_writer(writer) + .with_target(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_filter(request_dump_file_filter) + }); tracing_subscriber::registry() .with(stdout_layer) .with(file_layer) .with(recovery_file_layer) + .with(request_dump_file_layer) .try_init() .map_err(|_| LogError::AlreadyInitialized)?; Ok(LogGuard { _writers: guards }) @@ -131,8 +142,12 @@ fn application_file_filter(directive: &str) -> Result { let recovery_directive = format!("{OAUTH_RECOVERY_LOG_TARGET}=off") .parse() .expect("static OAuth recovery log directive is valid"); + let request_dump_directive = format!("{REQUEST_DUMP_LOG_TARGET}=off") + .parse() + .expect("static request dump log directive is valid"); EnvFilter::try_new(directive) .map(|filter| filter.add_directive(recovery_directive)) + .map(|filter| filter.add_directive(request_dump_directive)) .map_err(|_| LogError::InvalidFilter) } @@ -142,14 +157,41 @@ fn oauth_recovery_file_filter() -> EnvFilter { .expect("static OAuth recovery log filter is valid") } +fn request_dump_file_filter() -> EnvFilter { + format!("off,{REQUEST_DUMP_LOG_TARGET}=info") + .parse() + .expect("static request dump log filter is valid") +} + fn stdout_filter_directive(directive: &str, persistent_log_enabled: bool) -> String { if !persistent_log_enabled { - directive.to_owned() + format!("{directive},{REQUEST_DUMP_LOG_TARGET}=off") } else { "off,gateway_startup=info".to_owned() } } +fn rotating_file_writer( + config: &FileLoggingConfig, + file_prefix: &'static str, + thread_name: &'static str, +) -> Result<(NonBlocking, WorkerGuard), LogError> { + let maximum_bytes = config + .max_file_size_mb + .checked_mul(1024 * 1024) + .ok_or(LogError::SizeOverflow)?; + let writer = RotatingLogWriter::open( + config.directory.clone(), + file_prefix, + maximum_bytes, + config.retention_days, + config.max_files, + )?; + Ok(NonBlockingBuilder::default() + .thread_name(thread_name) + .finish(writer)) +} + struct RotatingLogWriter { directory: PathBuf, file_prefix: &'static str, diff --git a/backend/crates/gateway-host/tests/config.rs b/backend/crates/gateway-host/tests/config.rs index 396d15d48..273c23cd7 100644 --- a/backend/crates/gateway-host/tests/config.rs +++ b/backend/crates/gateway-host/tests/config.rs @@ -103,6 +103,7 @@ fn valid_config() -> HostConfig { max_file_size_mb: 100, max_files: 30, }, + request_dump: false, }, system_update, drain_timeout_seconds: 30, diff --git a/backend/crates/gateway-host/tests/logging.rs b/backend/crates/gateway-host/tests/logging.rs index bd17ce4b6..3f727751a 100644 --- a/backend/crates/gateway-host/tests/logging.rs +++ b/backend/crates/gateway-host/tests/logging.rs @@ -8,13 +8,18 @@ use gateway_host::system_update::SystemUpdateConfig; const LOG_DIRECTORY_ENV: &str = "CPR_LOGGING_TEST_DIRECTORY"; const CHILD_PROCESS_ENV: &str = "CPR_LOGGING_TEST_CHILD"; +const REQUEST_DUMP_ENABLED_ENV: &str = "CPR_LOGGING_TEST_REQUEST_DUMP_ENABLED"; const APPLICATION_LOG_FILE_PREFIX: &str = "codex-proxy-rs."; const OAUTH_RECOVERY_LOG_FILE_PREFIX: &str = "codex-proxy-rs-oath."; +const REQUEST_DUMP_LOG_FILE_PREFIX: &str = "codex-proxy-rs-request-dump."; const APPLICATION_LOG_TARGET: &str = "logging_test_application"; const APPLICATION_LOG_MARKER: &str = "application-file-filter-test"; const OAUTH_RECOVERY_LOG_TARGET: &str = "oauth_recovery"; const OAUTH_RECOVERY_LOG_MARKER: &str = "oauth-recovery-file-filter-test"; const OAUTH_RECOVERY_PROVIDER: &str = "openai"; +const REQUEST_DUMP_LOG_TARGET: &str = "request_dump"; +const REQUEST_DUMP_LOG_MARKER: &str = "request-dump-file-filter-test"; +const REQUEST_DUMP_SECRET: &str = "unredacted-authorization-value"; #[test] fn logging_requires_at_least_one_sink() { @@ -34,6 +39,7 @@ fn logging_requires_at_least_one_sink() { max_file_size_mb: 100, max_files: 30, }, + request_dump: false, }, system_update: SystemUpdateConfig::default(), drain_timeout_seconds: 30, @@ -48,11 +54,12 @@ fn logging_requires_at_least_one_sink() { } #[test] -fn oauth_recovery_file_logging_is_separate_and_overrides_global_log_level() { +fn sensitive_file_logging_is_separate_and_overrides_global_log_level() { if env::var_os(CHILD_PROCESS_ENV).is_some() { - write_oauth_recovery_log(PathBuf::from( - env::var_os(LOG_DIRECTORY_ENV).expect("child log directory"), - )); + write_sensitive_logs( + PathBuf::from(env::var_os(LOG_DIRECTORY_ENV).expect("child log directory")), + env::var(REQUEST_DUMP_ENABLED_ENV).is_ok_and(|value| value == "true"), + ); return; } @@ -60,14 +67,15 @@ fn oauth_recovery_file_logging_is_separate_and_overrides_global_log_level() { let output = Command::new(env::current_exe().expect("current test executable")) .args([ "--exact", - "logging::oauth_recovery_file_logging_is_separate_and_overrides_global_log_level", + "logging::sensitive_file_logging_is_separate_and_overrides_global_log_level", "--nocapture", ]) .env(CHILD_PROCESS_ENV, "1") .env(LOG_DIRECTORY_ENV, directory.path()) + .env(REQUEST_DUMP_ENABLED_ENV, "true") .env( "RUST_LOG", - "off,logging_test_application=info,oauth_recovery=off", + "off,logging_test_application=info,oauth_recovery=off,request_dump=trace", ) .output() .expect("run logging child process"); @@ -84,6 +92,8 @@ fn oauth_recovery_file_logging_is_separate_and_overrides_global_log_level() { assert!(application_log.contains(APPLICATION_LOG_MARKER)); assert!(!application_log.contains(&json_target_field(OAUTH_RECOVERY_LOG_TARGET))); assert!(!application_log.contains(OAUTH_RECOVERY_LOG_MARKER)); + assert!(!application_log.contains(&json_target_field(REQUEST_DUMP_LOG_TARGET))); + assert!(!application_log.contains(REQUEST_DUMP_SECRET)); let recovery_log = read_log_file_set(directory.path(), OAUTH_RECOVERY_LOG_FILE_PREFIX); assert!(recovery_log.contains(&json_target_field(OAUTH_RECOVERY_LOG_TARGET))); @@ -91,9 +101,49 @@ fn oauth_recovery_file_logging_is_separate_and_overrides_global_log_level() { assert!(!recovery_log.contains(&json_target_field(APPLICATION_LOG_TARGET))); assert!(!recovery_log.contains(APPLICATION_LOG_MARKER)); assert!(recovery_log.contains(&format!(r#""provider":"{OAUTH_RECOVERY_PROVIDER}""#))); + assert!(!recovery_log.contains(&json_target_field(REQUEST_DUMP_LOG_TARGET))); + assert!(!recovery_log.contains(REQUEST_DUMP_SECRET)); + + let request_dump_log = read_log_file_set(directory.path(), REQUEST_DUMP_LOG_FILE_PREFIX); + assert!(request_dump_log.contains(&json_target_field(REQUEST_DUMP_LOG_TARGET))); + assert!(request_dump_log.contains(REQUEST_DUMP_LOG_MARKER)); + assert!(request_dump_log.contains(REQUEST_DUMP_SECRET)); + assert!(!request_dump_log.contains(&json_target_field(APPLICATION_LOG_TARGET))); + assert!(!request_dump_log.contains(APPLICATION_LOG_MARKER)); + assert!(!request_dump_log.contains(&json_target_field(OAUTH_RECOVERY_LOG_TARGET))); + assert!(!request_dump_log.contains(OAUTH_RECOVERY_LOG_MARKER)); + + let disabled_directory = tempfile::tempdir().expect("create disabled log directory"); + let output = Command::new(env::current_exe().expect("current test executable")) + .args([ + "--exact", + "logging::sensitive_file_logging_is_separate_and_overrides_global_log_level", + "--nocapture", + ]) + .env(CHILD_PROCESS_ENV, "1") + .env(LOG_DIRECTORY_ENV, disabled_directory.path()) + .env(REQUEST_DUMP_ENABLED_ENV, "false") + .env( + "RUST_LOG", + "off,logging_test_application=info,oauth_recovery=off,request_dump=trace", + ) + .output() + .expect("run logging child process with request dump disabled"); + assert!( + output.status.success(), + "logging child process failed:\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + ); + assert!(!log_file_set_exists( + disabled_directory.path(), + REQUEST_DUMP_LOG_FILE_PREFIX + )); + let application_log = read_log_file_set(disabled_directory.path(), APPLICATION_LOG_FILE_PREFIX); + assert!(!application_log.contains(REQUEST_DUMP_SECRET)); } -fn write_oauth_recovery_log(directory: PathBuf) { +fn write_sensitive_logs(directory: PathBuf, request_dump: bool) { let config = HostConfig { listen: ListenConfig { host: "127.0.0.1".to_owned(), @@ -110,6 +160,7 @@ fn write_oauth_recovery_log(directory: PathBuf) { max_file_size_mb: 1, max_files: 1, }, + request_dump, }, system_update: SystemUpdateConfig::default(), drain_timeout_seconds: 30, @@ -134,6 +185,12 @@ fn write_oauth_recovery_log(directory: PathBuf) { marker = OAUTH_RECOVERY_LOG_MARKER, "OAuth recovery test record" ); + tracing::info!( + target: REQUEST_DUMP_LOG_TARGET, + marker = REQUEST_DUMP_LOG_MARKER, + authorization = REQUEST_DUMP_SECRET, + "request dump test record" + ); drop(bundle); } @@ -156,6 +213,18 @@ fn read_log_file_set(directory: &Path, file_prefix: &str) -> String { .collect() } +fn log_file_set_exists(directory: &Path, file_prefix: &str) -> bool { + fs::read_dir(directory) + .expect("read log directory") + .filter_map(Result::ok) + .any(|entry| { + entry + .file_name() + .to_str() + .is_some_and(|name| name.starts_with(file_prefix)) + }) +} + fn json_target_field(target: &str) -> String { format!(r#""target":"{target}""#) } diff --git a/backend/crates/providers/openai/src/transport/client_sse.rs b/backend/crates/providers/openai/src/transport/client_sse.rs index 7b81e1a8d..0f4cfb9da 100644 --- a/backend/crates/providers/openai/src/transport/client_sse.rs +++ b/backend/crates/providers/openai/src/transport/client_sse.rs @@ -37,10 +37,11 @@ use crate::transport::{ response_meta, websocket::{ CodexWebSocketConnection, CodexWebSocketExchangeError, CodexWebSocketPool, - CodexWebSocketPoolKey, CodexWebSocketStreamingExchange, DEFAULT_INITIAL_EVENT_TIMEOUT, - WEBSOCKET_FAST_PATH_BUDGET, WebSocketFastPath, WebSocketOriginBreaker, - WebSocketPoolDecision, execute_prepared_response_create_request_stream, - post_send_ambiguous, prepare_response_create_request_with_pool, websocket_audit_dir, + CodexWebSocketPoolKey, CodexWebSocketRequest, CodexWebSocketStreamingExchange, + DEFAULT_INITIAL_EVENT_TIMEOUT, WEBSOCKET_FAST_PATH_BUDGET, WebSocketFastPath, + WebSocketOriginBreaker, WebSocketPoolDecision, + execute_prepared_response_create_request_stream, post_send_ambiguous, + prepare_response_create_request_with_pool, websocket_audit_dir, write_websocket_audit_artifact_from_env, }, }; @@ -104,11 +105,26 @@ impl CodexBackendClient { upstream_body.insert("stream".to_owned(), serde_json::Value::Bool(true)); let body = serde_json::to_vec(&upstream_body).map_err(CodexClientError::RequestBodyEncode)?; + let endpoint = endpoint_url(&self.base_url, CODEX_RESPONSES_PATH); + tracing::info!( + target: "request_dump", + provider = "openai", + request_id = %context.request_id, + account_id = context.account_id.unwrap_or_default(), + request_direction = "upstream", + request_transport = "http_sse", + request_method = "POST", + request_url = %endpoint, + request_headers = ?headers, + request_body = %String::from_utf8_lossy(&body), + contains_sensitive_data = true, + "request dump (unredacted)" + ); let body = zstd::stream::encode_all(std::io::Cursor::new(body), 3) .map_err(CodexClientError::RequestCompression)?; let response = self .client - .post(endpoint_url(&self.base_url, CODEX_RESPONSES_PATH)) + .post(endpoint) .headers(headers) .header(CONTENT_ENCODING, HeaderValue::from_static("zstd")) .body(body) @@ -425,6 +441,7 @@ impl CodexBackendClient { prepared, } = *route; let delivery_wait_started_at = Instant::now(); + log_unredacted_websocket_request(&websocket_request, context); let mut exchange = match execute_prepared_response_create_request_stream( &websocket_request, prepared, @@ -495,6 +512,7 @@ impl CodexBackendClient { route: PreparedResponseRoute::WebSocket(route), .. }) => { + log_unredacted_websocket_request(&route.request, context); match execute_prepared_response_create_request_stream( &route.request, route.prepared, @@ -891,6 +909,26 @@ impl CodexBackendClient { } } +fn log_unredacted_websocket_request( + request: &CodexWebSocketRequest, + context: CodexRequestContext<'_>, +) { + tracing::info!( + target: "request_dump", + provider = "openai", + request_id = %context.request_id, + account_id = context.account_id.unwrap_or_default(), + request_direction = "upstream", + request_transport = "websocket", + request_method = "GET + response.create", + request_url = request.connection().endpoint(), + request_headers = ?request.connection().headers(), + request_payload = request.payload_text(), + contains_sensitive_data = true, + "request dump (unredacted)" + ); +} + /// 首个可投递帧前的交付边界结果。 enum DeliveryBoundary { /// 已越过边界，可开始向下游投递。 diff --git a/deploy/config.example.yaml b/deploy/config.example.yaml index 504ffdbac..05a0ac37f 100644 --- a/deploy/config.example.yaml +++ b/deploy/config.example.yaml @@ -20,6 +20,8 @@ host: logging: level: 'info' stdout: true + # 将完整请求头和正文写入独立文件；包含凭据，仅排障时开启。 + request_dump: false file: enabled: true # 相对路径以 deploy/config.yaml 所在目录为基准。