From 3fa9039bd70d2e4f6b4e614e11b971e261e604e2 Mon Sep 17 00:00:00 2001 From: jif Date: Mon, 14 Sep 2026 16:28:07 +0000 Subject: [PATCH] Label rollout compression failures by stage and I/O error kind (#45461) ## 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 --- codex-rs/rollout/src/compression.rs | 108 +++++++++++++----- .../rollout/src/compression/error_metrics.rs | 53 +++++++++ 2 files changed, 132 insertions(+), 29 deletions(-) create mode 100644 codex-rs/rollout/src/compression/error_metrics.rs diff --git a/codex-rs/rollout/src/compression.rs b/codex-rs/rollout/src/compression.rs index d71d56c804..996ac4a7e0 100644 --- a/codex-rs/rollout/src/compression.rs +++ b/codex-rs/rollout/src/compression.rs @@ -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, thread_id: codex_protocol::ThreadId, ) -> io::Result { - 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)]); diff --git a/codex-rs/rollout/src/compression/error_metrics.rs b/codex-rs/rollout/src/compression/error_metrics.rs new file mode 100644 index 0000000000..3c75c3bd91 --- /dev/null +++ b/codex-rs/rollout/src/compression/error_metrics.rs @@ -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), + ], + ); + } +}