starred/codex-proxy-rs
1
0
Fork 0
mirror of https://github.com/zyycn/codex-proxy-rs.git synced 2026-11-06 08:06:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(logging): add opt-in request dumps

Browse Source 
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 <noreply@openai.com>
This commit is contained in:
Zyy
2026-09-04 19:15:28 +08:00
co-authored by Codex
parent 51e9981bee
commit 8f242e2875
8 changed files with 244 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/crates/gateway-api/src/openai/responses/http.rs
+13
View File
@@ -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;
backend/crates/gateway-api/src/openai/responses/websocket/mod.rs
+38 -2
View File
@@ -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<IpAddr>,
user_agent: Option<String>,
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::<str>::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
backend/crates/gateway-host/src/config.rs
+3
View File
@@ -118,6 +118,9 @@ pub struct LoggingConfig {
pub level: String,
pub stdout: bool,
pub file: FileLoggingConfig,
/// 将完整请求原文写入独立诊断日志；默认关闭，因为内容包含凭据。
#[serde(default)]
pub request_dump: bool,
}
impl LoggingConfig {
backend/crates/gateway-host/src/logging.rs
+68 -26
View File
@@ -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<LogGuard, LogError>
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<LogGuard, LogError>
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<LogGuard, LogError>
.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<EnvFilter, LogError> {
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,
backend/crates/gateway-host/tests/config.rs
+1
View File
@@ -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,
backend/crates/gateway-host/tests/logging.rs
+76 -7
View File
@@ -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}""#)
}
backend/crates/providers/openai/src/transport/client_sse.rs
+43 -5
View File
@@ -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 {
/// 已越过边界，可开始向下游投递。
deploy/config.example.yaml
+2
View File
@@ -20,6 +20,8 @@ host:
logging:
level: 'info'
stdout: true
# 将完整请求头和正文写入独立文件；包含凭据，仅排障时开启。
request_dump: false
file:
enabled: true
# 相对路径以 deploy/config.yaml 所在目录为基准。