openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-15 12:08:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

Label rollout compression failures by stage and I/O error kind (#45461)

Browse Source 
## Why

Rollout compression failure counters report only that an operation failed,
without identifying the failing stage or I/O error kind.

## What changed

- Add `stage` and `error_kind` labels to failure counters for compression runs,
  individual files, materialization for append, and stale temporary file cleanup.
- Record failures at their source, including lock acquisition and task joins,
  and avoid counting file compression failures twice.
- Use static stage labels and a fixed set of error categories, keeping error
  messages, paths, and rollout contents out of metric tags.

GitOrigin-RevId: ac2bfc7ae4cec4e9f60de9557b03345a3986aa53
This commit is contained in:
jif
2026-09-14 16:28:07 +00:00
committed by copyberry
parent 374c4b2d82
commit 3fa9039bd7
2 changed files with 132 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

108
codex-rs/rollout/src/compression.rs
View File

@@ -15,6 +15,10 @@ use std::os::unix::fs::OpenOptionsExt;
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
mod error_metrics;
use error_metrics::FailureMetric;
const COMPRESSED_SUFFIX: &str = ".zst";
const MAX_NOT_FOUND_RETRIES: usize = 3;
const OPEN_ROLLOUT_LINE_READER_RETRY_DELAY: Duration = Duration::from_millis(50);
@@ -72,7 +76,8 @@ pub(crate) async fn materialize_rollout_for_append(
materialize_rollout_for_append_blocking(path.as_path())
})
.await
.map_err(io::Error::other)?
.map_err(io::Error::other)
.inspect_err(|err| FailureMetric::Materialize.record("task_join", err))?
}
/// Materializes a compressed rollout back to plain `.jsonl` for blocking append paths.
@@ -90,28 +95,39 @@ pub(crate) fn materialize_rollout_for_append_blocking(path: &Path) -> io::Result
let temp_path = temp_path_for(plain_path.as_path(), "decompress");
if let Some(parent) = plain_path.parent() {
std::fs::create_dir_all(parent)?;
std::fs::create_dir_all(parent)
.inspect_err(|err| FailureMetric::Materialize.record("prepare_directory", err))?;
}
let mut stage = "read_metadata";
let result: io::Result<()> = (|| {
let metadata = std::fs::metadata(compressed_path.as_path())?;
let permissions = metadata.permissions();
stage = "create_temp";
let mut output = create_file_with_permissions(temp_path.as_path(), &permissions)?;
{
stage = "open_source";
let input = File::open(compressed_path.as_path())?;
stage = "decode_and_write";
let mut decoder = zstd::stream::read::Decoder::new(input)?;
io::copy(&mut decoder, &mut output)?;
}
stage = "flush";
output.flush()?;
stage = "sync";
output.sync_all()?;
stage = "publish";
match std::fs::hard_link(temp_path.as_path(), plain_path.as_path()) {
Ok(()) => {}
Err(err) if err.kind() == io::ErrorKind::AlreadyExists => {}
Err(_) => persist_temp_file_noclobber(temp_path.as_path(), plain_path.as_path())?,
}
stage = "set_metadata";
output.set_times(std::fs::FileTimes::new().set_modified(metadata.modified()?))?;
stage = "sync";
output.sync_all()?;
drop(output);
let _ = std::fs::remove_file(temp_path.as_path());
stage = "remove_source";
match std::fs::remove_file(compressed_path.as_path()) {
Ok(()) => {}
Err(err) if err.kind() == io::ErrorKind::NotFound => {}
@@ -119,9 +135,9 @@ pub(crate) fn materialize_rollout_for_append_blocking(path: &Path) -> io::Result
}
Ok(())
})();
if result.is_err() {
if let Err(err) = &result {
let _ = std::fs::remove_file(temp_path.as_path());
metrics::materialize("failed");
FailureMetric::Materialize.record(stage, err);
}
result?;
metrics::materialize("decompressed");
@@ -254,6 +270,7 @@ mod worker {
use crate::SESSIONS_SUBDIR;
use super::RolloutFile;
use super::error_metrics::FailureMetric;
use super::metrics;
use super::path;
@@ -351,7 +368,8 @@ mod worker {
pub(super) async fn run(codex_home: PathBuf) -> io::Result<()> {
let Some(_maintenance_guard) =
crate::try_acquire_rollout_maintenance_lock(codex_home.as_path())?
crate::try_acquire_rollout_maintenance_lock(codex_home.as_path())
.inspect_err(|err| FailureMetric::Run.record("maintenance_lock", err))?
else {
metrics::run("skipped_maintenance");
debug!(
@@ -371,7 +389,7 @@ mod worker {
return Ok(());
}
Err(err) => {
metrics::run("failed");
FailureMetric::Run.record("run_marker", &err);
return Err(err);
}
};
@@ -379,8 +397,10 @@ mod worker {
metrics::run("started");
let started_at = Instant::now();
let writer_locks = Arc::new(crate::WriterLockCoordinator::new(&codex_home));
let mut stage = "temp_cleanup";
let result = async {
cleanup_stale_temps(codex_home.as_path()).await?;
stage = "scan";
let mut stats = CompressionStats::default();
for root in [
codex_home.join(ARCHIVED_SESSIONS_SUBDIR),
@@ -398,7 +418,7 @@ mod worker {
let stats = match result {
Ok(stats) => stats,
Err(err) => {
metrics::run("failed");
FailureMetric::Run.record(stage, &err);
metrics::run_duration("failed", started_at.elapsed());
return Err(err);
}
@@ -613,14 +633,14 @@ mod worker {
}
Ok((path, duration, Err(err))) => {
stats.failed = stats.failed.saturating_add(1);
metrics::file("failed");
// The failing operation records its stage before returning the error.
metrics::file_duration("failed", duration);
warn!("failed to compress rollout {}: {err}", path.display());
}
Err(err) => {
stats.failed = stats.failed.saturating_add(1);
metrics::file("failed");
warn!("rollout compression task failed: {err}");
FailureMetric::File.record("task_join", &io::Error::other(err));
}
}
}
@@ -630,7 +650,9 @@ mod worker {
writer_locks: &Arc<crate::WriterLockCoordinator>,
thread_id: codex_protocol::ThreadId,
) -> io::Result<CompressionMeasurement> {
let before = match cold_file_state(path)? {
let before = match cold_file_state(path)
.inspect_err(|err| FailureMetric::File.record("read_metadata", err))?
{
ColdFileState::Cold(state) => state,
ColdFileState::NotCold(state) => {
return Ok(CompressionMeasurement::new(
@@ -654,35 +676,57 @@ mod worker {
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
.unwrap_or_else(|| Path::new("."));
std::fs::create_dir_all(temp_dir)?;
std::fs::create_dir_all(temp_dir)
.inspect_err(|err| FailureMetric::File.record("prepare_directory", err))?;
let mut temp_file = tempfile::Builder::new()
.prefix("rollout-compress-")
.suffix(TEMP_SUFFIX)
.tempfile_in(temp_dir)?;
encode_zstd_to_writer(path, temp_file.as_file_mut())?;
temp_file.as_file_mut().flush()?;
verify_zstd(temp_file.path())?;
if !same_file_state(path, &before)? {
.tempfile_in(temp_dir)
.inspect_err(|err| FailureMetric::File.record("create_temp", err))?;
encode_zstd_to_writer(path, temp_file.as_file_mut())
.inspect_err(|err| FailureMetric::File.record("encode_and_write", err))?;
temp_file
.as_file_mut()
.flush()
.inspect_err(|err| FailureMetric::File.record("flush", err))?;
verify_zstd(temp_file.path())
.inspect_err(|err| FailureMetric::File.record("verify", err))?;
if !same_file_state(path, &before)
.inspect_err(|err| FailureMetric::File.record("recheck_source", err))?
{
return Ok(CompressionMeasurement::new(
CompressionOutcome::SkippedChanged,
source_bytes,
/*compressed_bytes*/ None,
));
}
set_file_metadata(temp_file.as_file(), before.modified, &before.permissions)?;
temp_file.as_file().sync_all()?;
let compressed_bytes = temp_file.as_file().metadata()?.len();
set_file_metadata(temp_file.as_file(), before.modified, &before.permissions)
.inspect_err(|err| FailureMetric::File.record("set_metadata", err))?;
temp_file
.as_file()
.sync_all()
.inspect_err(|err| FailureMetric::File.record("sync", err))?;
let compressed_bytes = temp_file
.as_file()
.metadata()
.inspect_err(|err| FailureMetric::File.record("read_metadata", err))?
.len();
// Encoding and verification do not block writers. Coordination prevents writer
// acquisition while we recheck, publish, and remove the original file.
let Some(_publication_guard) = writer_locks.try_acquire_for_publication(thread_id)? else {
let Some(_publication_guard) = writer_locks
.try_acquire_for_publication(thread_id)
.inspect_err(|err| FailureMetric::File.record("writer_lock", err))?
else {
return Ok(CompressionMeasurement::new(
CompressionOutcome::SkippedBusy,
source_bytes,
/*compressed_bytes*/ None,
));
};
if !same_file_state(path, &before)? {
if !same_file_state(path, &before)
.inspect_err(|err| FailureMetric::File.record("recheck_source", err))?
{
return Ok(CompressionMeasurement::new(
CompressionOutcome::SkippedChanged,
source_bytes,
@@ -699,9 +743,14 @@ mod worker {
/*compressed_bytes*/ None,
));
}
Err(err) => return Err(err.error),
Err(err) => {
FailureMetric::File.record("publish", &err.error);
return Err(err.error);
}
}
if !same_file_state(path, &before)? {
if !same_file_state(path, &before)
.inspect_err(|err| FailureMetric::File.record("recheck_source", err))?
{
let _ = std::fs::remove_file(compressed_path.as_path());
return Ok(CompressionMeasurement::new(
CompressionOutcome::SkippedChanged,
@@ -709,7 +758,8 @@ mod worker {
/*compressed_bytes*/ None,
));
}
std::fs::remove_file(path)?;
std::fs::remove_file(path)
.inspect_err(|err| FailureMetric::File.record("remove_source", err))?;
Ok(CompressionMeasurement::new(
CompressionOutcome::Compressed,
source_bytes,
@@ -848,7 +898,7 @@ mod worker {
Ok(()) => metrics::temp_cleanup("removed"),
Err(err) if err.kind() == io::ErrorKind::NotFound => {}
Err(err) => {
metrics::temp_cleanup("failed");
FailureMetric::TempCleanup.record("remove_temp", &err);
warn!(
"failed to remove stale rollout temp {}: {err}",
path.display()
@@ -866,16 +916,16 @@ mod metrics {
use std::time::Duration;
const FILE_COMPRESSED_BYTES_HISTOGRAM: &str = "codex.rollout_compression.file.compressed_bytes";
const FILE_COUNTER: &str = "codex.rollout_compression.file";
pub(super) const FILE_COUNTER: &str = "codex.rollout_compression.file";
const FILE_DURATION_HISTOGRAM: &str = "codex.rollout_compression.file.duration_ms";
const FILE_SOURCE_BYTES_HISTOGRAM: &str = "codex.rollout_compression.file.source_bytes";
const FILE_COMPRESSION_RATIO_HISTOGRAM: &str =
"codex.rollout_compression.file.compression_ratio";
const MATERIALIZE_COUNTER: &str = "codex.rollout_compression.materialize";
const RUN_COUNTER: &str = "codex.rollout_compression.run";
pub(super) const MATERIALIZE_COUNTER: &str = "codex.rollout_compression.materialize";
pub(super) const RUN_COUNTER: &str = "codex.rollout_compression.run";
const RUN_DURATION_HISTOGRAM: &str = "codex.rollout_compression.run.duration_ms";
const RATIO_BASIS_POINTS: u128 = 10_000;
const TEMP_CLEANUP_COUNTER: &str = "codex.rollout_compression.temp_cleanup";
pub(super) const TEMP_CLEANUP_COUNTER: &str = "codex.rollout_compression.temp_cleanup";
pub(super) fn file(outcome: &'static str) {
counter(FILE_COUNTER, &[("outcome", outcome)]);

53
codex-rs/rollout/src/compression/error_metrics.rs Normal file
View File

@@ -0,0 +1,53 @@
//! Failure labels for rollout compression. Only static stages and bounded I/O kinds
//! are exported; error messages, paths, and rollout contents never become metric tags.
use std::io;
use super::metrics;
pub(super) enum FailureMetric {
File,
Materialize,
Run,
TempCleanup,
}
impl FailureMetric {
pub(super) fn record(self, stage: &'static str, error: &io::Error) {
let (name, outcome_key) = match self {
Self::File => (metrics::FILE_COUNTER, "outcome"),
Self::Materialize => (metrics::MATERIALIZE_COUNTER, "outcome"),
Self::Run => (metrics::RUN_COUNTER, "status"),
Self::TempCleanup => (metrics::TEMP_CLEANUP_COUNTER, "outcome"),
};
let error_kind = match error.kind() {
io::ErrorKind::NotFound => "not_found",
io::ErrorKind::PermissionDenied => "permission_denied",
io::ErrorKind::AlreadyExists => "already_exists",
io::ErrorKind::InvalidInput => "invalid_input",
io::ErrorKind::InvalidData => "invalid_data",
io::ErrorKind::TimedOut => "timed_out",
io::ErrorKind::WriteZero => "write_zero",
io::ErrorKind::Interrupted => "interrupted",
io::ErrorKind::Unsupported => "unsupported",
io::ErrorKind::UnexpectedEof => "unexpected_eof",
io::ErrorKind::StorageFull => "storage_full",
io::ErrorKind::ReadOnlyFilesystem => "read_only_filesystem",
io::ErrorKind::NotADirectory => "not_a_directory",
io::ErrorKind::IsADirectory => "is_a_directory",
_ => "other",
};
let Some(metrics) = codex_otel::global() else {
return;
};
let _ = metrics.counter(
name,
/*inc*/ 1,
&[
(outcome_key, "failed"),
("stage", stage),
("error_kind", error_kind),
],
);
}
}