From 9f4fac8ec4e74318ae272afedd118fc635a1023d Mon Sep 17 00:00:00 2001 From: jif-oai Date: Mon, 1 Jun 2026 22:26:32 +0200 Subject: [PATCH] Add rollout compression counters (#25679) ## Summary Add counter telemetry for the local rollout compression worker so we can see when it runs, why it skips, and how individual file/materialization paths resolve. ## Changes - Emit `codex.rollout_compression.run` with statuses for start, completion, failure, duplicate-run skip, and missing runtime skip. - Emit `codex.rollout_compression.file` outcomes for scanned, compressed, skipped, and failed compression candidates. - Emit `codex.rollout_compression.temp_cleanup` and `codex.rollout_compression.materialize` counters for cleanup and decompression paths. ## Validation - `just fmt` - `just test -p codex-rollout` - `just fix -p codex-rollout` --- codex-rs/rollout/src/compression.rs | 142 +++++++++++++++++++++++----- 1 file changed, 116 insertions(+), 26 deletions(-) diff --git a/codex-rs/rollout/src/compression.rs b/codex-rs/rollout/src/compression.rs index 8211f79d6..99e4c6b4b 100644 --- a/codex-rs/rollout/src/compression.rs +++ b/codex-rs/rollout/src/compression.rs @@ -75,10 +75,12 @@ pub(crate) async fn materialize_rollout_for_append(path: &Path) -> io::Result io::Result { let plain_path = plain_rollout_path(path); if plain_path.exists() { + metrics::materialize("plain_exists"); return Ok(plain_path); } let compressed_path = path::compressed_rollout_path(plain_path.as_path()); if !compressed_path.exists() { + metrics::materialize("missing"); return Ok(plain_path); } @@ -111,8 +113,10 @@ pub(crate) fn materialize_rollout_for_append_blocking(path: &Path) -> io::Result })(); if result.is_err() { let _ = std::fs::remove_file(temp_path.as_path()); + metrics::materialize("failed"); } result?; + metrics::materialize("decompressed"); Ok(plain_path) } @@ -241,6 +245,7 @@ mod worker { use crate::SESSIONS_SUBDIR; use super::RolloutFile; + use super::metrics; use super::path; const TEMP_SUFFIX: &str = ".tmp"; @@ -318,6 +323,7 @@ mod worker { pub(super) fn spawn(codex_home: PathBuf) { let Ok(handle) = tokio::runtime::Handle::try_current() else { + metrics::run("skipped_no_runtime"); warn!( "failed to start rollout compression worker for {}: no Tokio runtime", codex_home.display() @@ -335,25 +341,46 @@ mod worker { } pub(super) async fn run(codex_home: PathBuf) -> io::Result<()> { - let Some(marker) = CompressionRunMarker::try_claim(codex_home.as_path())? else { - debug!( - "rollout compression worker recently ran or is already running for {}", - codex_home.display() - ); - return Ok(()); + let marker = match CompressionRunMarker::try_claim(codex_home.as_path()) { + Ok(Some(marker)) => marker, + Ok(None) => { + metrics::run("skipped_already_running"); + debug!( + "rollout compression worker recently ran or is already running for {}", + codex_home.display() + ); + return Ok(()); + } + Err(err) => { + metrics::run("failed"); + return Err(err); + } }; + metrics::run("started"); let started_at = Instant::now(); - cleanup_stale_temps(codex_home.as_path()).await?; - let mut stats = CompressionStats::default(); - if started_at.elapsed() < WORKER_MAX_RUNTIME { - let archived_root = codex_home.join(ARCHIVED_SESSIONS_SUBDIR); - compress_rollouts_in_root(archived_root.as_path(), started_at, &mut stats).await?; + let result = async { + cleanup_stale_temps(codex_home.as_path()).await?; + let mut stats = CompressionStats::default(); + if started_at.elapsed() < WORKER_MAX_RUNTIME { + let archived_root = codex_home.join(ARCHIVED_SESSIONS_SUBDIR); + compress_rollouts_in_root(archived_root.as_path(), started_at, &mut stats).await?; + } + Ok::<_, io::Error>(stats) } + .await; + let stats = match result { + Ok(stats) => stats, + Err(err) => { + metrics::run("failed"); + return Err(err); + } + }; info!( "rollout compression worker finished: scanned={}, compressed={}, skipped={}, failed={}", stats.scanned, stats.compressed, stats.skipped, stats.failed ); + metrics::run("completed"); marker.persist(); Ok(()) } @@ -434,6 +461,7 @@ mod worker { } let path = rollout_file.into_path(); stats.scanned = stats.scanned.saturating_add(1); + metrics::file("scanned"); while jobs.len() >= MAX_CONCURRENT_COMPRESSION_JOBS { collect_next_compression_job(&mut jobs, stats).await; } @@ -447,7 +475,26 @@ mod worker { Ok(()) } - type CompressionJobResult = (PathBuf, io::Result); + type CompressionJobResult = (PathBuf, io::Result); + + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + enum CompressionOutcome { + Compressed, + SkippedNotCold, + SkippedChanged, + SkippedAlreadyCompressed, + } + + impl CompressionOutcome { + fn tag(self) -> &'static str { + match self { + CompressionOutcome::Compressed => "compressed", + CompressionOutcome::SkippedNotCold => "skipped_not_cold", + CompressionOutcome::SkippedChanged => "skipped_changed", + CompressionOutcome::SkippedAlreadyCompressed => "skipped_already_compressed", + } + } + } async fn drain_compression_jobs( jobs: &mut JoinSet, @@ -466,27 +513,35 @@ mod worker { return; }; match result { - Ok((_, Ok(true))) => stats.compressed = stats.compressed.saturating_add(1), - Ok((_, Ok(false))) => stats.skipped = stats.skipped.saturating_add(1), + Ok((_, Ok(CompressionOutcome::Compressed))) => { + stats.compressed = stats.compressed.saturating_add(1); + metrics::file(CompressionOutcome::Compressed.tag()); + } + Ok((_, Ok(outcome))) => { + stats.skipped = stats.skipped.saturating_add(1); + metrics::file(outcome.tag()); + } Ok((path, Err(err))) => { stats.failed = stats.failed.saturating_add(1); + metrics::file("failed"); 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}"); } } } - fn compress_rollout_if_cold_blocking(path: &Path) -> io::Result { + fn compress_rollout_if_cold_blocking(path: &Path) -> io::Result { let before = match cold_file_state(path)? { Some(state) => state, - None => return Ok(false), + None => return Ok(CompressionOutcome::SkippedNotCold), }; let compressed_path = path::compressed_rollout_path(path); if compressed_path.exists() { - return Ok(false); + return Ok(CompressionOutcome::SkippedAlreadyCompressed); } let temp_dir = compressed_path @@ -502,22 +557,24 @@ mod worker { temp_file.as_file_mut().flush()?; verify_zstd(temp_file.path())?; if !same_file_state(path, &before)? { - return Ok(false); + return Ok(CompressionOutcome::SkippedChanged); } set_file_metadata(temp_file.as_file(), before.modified, &before.permissions)?; temp_file.as_file().sync_all()?; match temp_file.persist_noclobber(compressed_path.as_path()) { Ok(_) => {} - Err(err) if err.error.kind() == io::ErrorKind::AlreadyExists => return Ok(false), + Err(err) if err.error.kind() == io::ErrorKind::AlreadyExists => { + return Ok(CompressionOutcome::SkippedAlreadyCompressed); + } Err(err) => return Err(err.error), } if !same_file_state(path, &before)? { let _ = std::fs::remove_file(compressed_path.as_path()); - return Ok(false); + return Ok(CompressionOutcome::SkippedChanged); } std::fs::remove_file(path)?; - Ok(true) + Ok(CompressionOutcome::Compressed) } struct FileState { @@ -643,12 +700,15 @@ mod worker { continue; } match tokio::fs::remove_file(path.as_path()).await { - Ok(()) => {} + Ok(()) => metrics::temp_cleanup("removed"), Err(err) if err.kind() == io::ErrorKind::NotFound => {} - Err(err) => warn!( - "failed to remove stale rollout temp {}: {err}", - path.display() - ), + Err(err) => { + metrics::temp_cleanup("failed"); + warn!( + "failed to remove stale rollout temp {}: {err}", + path.display() + ); + } } } } @@ -657,6 +717,36 @@ mod worker { } } +mod metrics { + const FILE_COUNTER: &str = "codex.rollout_compression.file"; + const MATERIALIZE_COUNTER: &str = "codex.rollout_compression.materialize"; + const RUN_COUNTER: &str = "codex.rollout_compression.run"; + const TEMP_CLEANUP_COUNTER: &str = "codex.rollout_compression.temp_cleanup"; + + pub(super) fn file(outcome: &'static str) { + counter(FILE_COUNTER, &[("outcome", outcome)]); + } + + pub(super) fn materialize(outcome: &'static str) { + counter(MATERIALIZE_COUNTER, &[("outcome", outcome)]); + } + + pub(super) fn run(status: &'static str) { + counter(RUN_COUNTER, &[("status", status)]); + } + + pub(super) fn temp_cleanup(outcome: &'static str) { + counter(TEMP_CLEANUP_COUNTER, &[("outcome", outcome)]); + } + + fn counter(name: &str, tags: &[(&str, &str)]) { + let Some(metrics) = codex_otel::global() else { + return; + }; + let _ = metrics.counter(name, /*inc*/ 1, tags); + } +} + /// Returns the existing rollout path, preferring the plain `.jsonl` file over /// its `.jsonl.zst` compressed sibling. pub async fn existing_rollout_path(path: &Path) -> Option {