diff --git a/Cargo.lock b/Cargo.lock index 098244fbb..d2f4d4343 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9593,6 +9593,7 @@ dependencies = [ "futures-util", "glob", "jiff", + "libc", "metrics", "num_cpus", "nvml-wrapper", diff --git a/crates/obs/Cargo.toml b/crates/obs/Cargo.toml index ff6b98e85..454305bed 100644 --- a/crates/obs/Cargo.toml +++ b/crates/obs/Cargo.toml @@ -110,6 +110,9 @@ nvml-wrapper = { workspace = true, optional = true } [target.'cfg(any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))'.dependencies] pyroscope = { workspace = true, features = ["backend-pprof-rs"], optional = true } +[target.'cfg(unix)'.dependencies] +libc = { workspace = true } + [dev-dependencies] tokio = { workspace = true, features = ["full"] } diff --git a/crates/obs/src/cleaner/compress.rs b/crates/obs/src/cleaner/compress.rs index bae28bb76..b38f19391 100644 --- a/crates/obs/src/cleaner/compress.rs +++ b/crates/obs/src/cleaner/compress.rs @@ -35,6 +35,11 @@ const LOG_COMPONENT_OBS: &str = "obs"; const LOG_SUBSYSTEM_LOG_CLEANER: &str = "log_cleaner"; const EVENT_LOG_CLEANER_COMPRESSION_STATE: &str = "log_cleaner_compression_state"; +/// Conservative archive-size fraction used only for dry-run freed-byte +/// projection. Chosen to under-estimate reclaim (never overstate it) since the +/// true compression ratio is unknown until a real run. +const DRY_RUN_ESTIMATED_ARCHIVE_RATIO: f64 = 0.5; + /// Compression options shared by serial and parallel cleaner paths. /// /// The core cleaner prepares this immutable bundle once per cleanup pass and @@ -43,9 +48,9 @@ const EVENT_LOG_CLEANER_COMPRESSION_STATE: &str = "log_cleaner_compression_state pub(super) struct CompressionOptions { /// Preferred compression codec for the current cleanup pass. pub algorithm: CompressionAlgorithm, - /// Gzip level (1..=9). + /// Gzip level (0..=9; 0 = store / no compression). pub gzip_level: u32, - /// Zstd level (1..=21). + /// Zstd level (0..=22; 0 = codec default). pub zstd_level: i32, /// Internal zstd worker threads used by zstdmt. pub zstd_workers: usize, @@ -101,11 +106,14 @@ pub(super) fn compress_file(path: &Path, options: &CompressionOptions) -> Result fn compress_gzip(path: &Path, level: u32, dry_run: bool) -> Result { let archive_path = archive_path(path, CompressionAlgorithm::Gzip); compress_with_writer(path, &archive_path, dry_run, CompressionAlgorithm::Gzip, |reader, writer| { - let mut encoder = GzEncoder::new(writer, Compression::new(level.clamp(1, 9))); + let mut encoder = GzEncoder::new(writer, Compression::new(level.clamp(0, 9))); let written = std::io::copy(reader, &mut encoder)?; let mut out = encoder.finish()?; out.flush()?; - Ok(written) + // Hand the underlying file back so the caller can `sync_all()` the + // archive data before renaming and deleting the source. + let file = out.into_inner().map_err(|e| e.into_error())?; + Ok((written, file)) }) } @@ -113,15 +121,109 @@ fn compress_gzip(path: &Path, level: u32, dry_run: bool) -> Result Result { let archive_path = archive_path(path, CompressionAlgorithm::Zstd); compress_with_writer(path, &archive_path, dry_run, CompressionAlgorithm::Zstd, |reader, writer| { - let mut encoder = zstd::stream::Encoder::new(writer, level.clamp(1, 21))?; + let mut encoder = zstd::stream::Encoder::new(writer, level.clamp(0, 22))?; encoder.multithread(workers.max(1) as u32)?; let written = std::io::copy(reader, &mut encoder)?; let mut out = encoder.finish()?; out.flush()?; - Ok(written) + // Hand the underlying file back so the caller can `sync_all()` the + // archive data before renaming and deleting the source. + let file = out.into_inner().map_err(|e| e.into_error())?; + Ok((written, file)) }) } +/// RAII guard that removes a temporary archive on drop unless disarmed. +/// +/// This makes the compression path panic-safe: an early return *or a panic* +/// between creating the temp file and the final rename still cleans up the +/// partial artifact instead of leaking a `*.tmp` orphan. +struct TmpFileGuard { + path: Option, +} + +impl TmpFileGuard { + fn new(path: PathBuf) -> Self { + Self { path: Some(path) } + } + + fn disarm(&mut self) { + self.path = None; + } +} + +impl Drop for TmpFileGuard { + fn drop(&mut self) { + if let Some(path) = self.path.take() { + let _ = std::fs::remove_file(&path); + } + } +} + +/// Create the temporary archive file, refusing to follow or overwrite a +/// symlink planted at the temp path. +/// +/// `create_new` (`O_CREAT|O_EXCL`) rejects a pre-existing entry, and on Unix +/// `O_NOFOLLOW` additionally refuses a symlink at the final component. Without +/// these, an attacker with write access to the log directory could pre-plant +/// `.tmp` as a symlink and have the compressor truncate/overwrite an +/// arbitrary external file. The deletion path already refuses symlinks; this +/// closes the same gap on the compression path. +fn create_tmp_archive(path: &Path, source_mode: Option) -> std::io::Result { + let mut opts = std::fs::OpenOptions::new(); + opts.write(true).create_new(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + opts.custom_flags(libc::O_NOFOLLOW); + // Restrictive from creation: the temp file holds the full plaintext of + // a possibly-0600 audit log, so never expose it wider than the source. + opts.mode(source_mode.unwrap_or(0o600)); + } + let _ = source_mode; + opts.open(path) +} + +/// Cheap sanity check that an existing archive starts with the expected codec +/// magic bytes, so a zero/truncated/foreign file is not trusted as a valid +/// prior result. Only called on a confirmed regular, non-empty file. +fn archive_header_ok(path: &Path, algorithm: CompressionAlgorithm) -> bool { + use std::io::Read; + let mut file = match File::open(path) { + Ok(file) => file, + Err(_) => return false, + }; + let mut buf = [0u8; 4]; + let read = match file.read(&mut buf) { + Ok(read) => read, + Err(_) => return false, + }; + match algorithm { + // gzip: 0x1f 0x8b + CompressionAlgorithm::Gzip => read >= 2 && buf[0] == 0x1f && buf[1] == 0x8b, + // zstd frame magic: 0x28 0xb5 0x2f 0xfd (little-endian 0xFD2FB528) + CompressionAlgorithm::Zstd => read >= 4 && buf == [0x28, 0xb5, 0x2f, 0xfd], + } +} + +/// Best-effort fsync of the directory containing `path`. +/// +/// Persisting the directory entry makes a rename (or, for the deletion path, an +/// unlink) of an entry within it durable across a crash. Opening a directory as +/// a file is not portable (notably on Windows), so any failure is ignored. +fn sync_parent_dir(path: &Path) { + if let Some(parent) = path.parent() { + let dir = if parent.as_os_str().is_empty() { + Path::new(".") + } else { + parent + }; + if let Ok(handle) = File::open(dir) { + let _ = handle.sync_all(); + } + } +} + fn compress_with_writer( path: &Path, archive_path: &Path, @@ -130,30 +232,50 @@ fn compress_with_writer( mut writer_fn: F, ) -> Result where - F: FnMut(&mut BufReader, BufWriter) -> Result, + F: FnMut(&mut BufReader, BufWriter) -> Result<(u64, File), std::io::Error>, { - // Keep idempotent behavior: existing archive means this file has already - // been handled in a previous cleanup pass. - if archive_path.exists() { - debug!(event = EVENT_LOG_CLEANER_COMPRESSION_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_LOG_CLEANER, file = ?archive_path, state = "archive_exists", "log cleaner compression state changed"); - let input_bytes = std::fs::metadata(path).map(|m| m.len()).unwrap_or(0); - let output_bytes = std::fs::metadata(archive_path).map(|m| m.len()).unwrap_or(0); - return Ok(CompressionOutput { - archive_path: archive_path.to_path_buf(), - algorithm_used, - input_bytes, - output_bytes, - }); + // Idempotency: an existing *complete* archive means a previous pass already + // handled this file, so we can skip recompression and let the caller delete + // the source. But we must not trust just any entry at `archive_path`: + // * `Path::exists()` follows symlinks, so a planted `.gz` symlink + // would green-light deleting the source with no real archive; + // * a zero-length/truncated archive left by a crashed run (see OLC-01) + // would be trusted, then the source deleted — unrecoverable loss. + // Use symlink_metadata (no follow) and require a regular, non-empty file + // with a valid codec header before trusting it. Anything else falls through + // to recompression, whose atomic create_new+rename replaces the bad entry + // (a symlink is replaced, never followed or deleted through). + match std::fs::symlink_metadata(archive_path) { + Ok(meta) if meta.file_type().is_file() && meta.len() > 0 && archive_header_ok(archive_path, algorithm_used) => { + debug!(event = EVENT_LOG_CLEANER_COMPRESSION_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_LOG_CLEANER, file = ?archive_path, state = "archive_exists", "log cleaner compression state changed"); + let input_bytes = std::fs::metadata(path).map(|m| m.len()).unwrap_or(0); + return Ok(CompressionOutput { + archive_path: archive_path.to_path_buf(), + algorithm_used, + input_bytes, + output_bytes: meta.len(), + }); + } + Ok(_) => { + warn!(event = EVENT_LOG_CLEANER_COMPRESSION_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_LOG_CLEANER, file = ?archive_path, result = "archive_untrusted_recompressing", "log cleaner compression state changed"); + } + Err(_) => {} } - let input_bytes = std::fs::metadata(path).map(|m| m.len()).unwrap_or(0); if dry_run { - info!(event = EVENT_LOG_CLEANER_COMPRESSION_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_LOG_CLEANER, state = "dry_run_compress", file = ?path, archive = ?archive_path, input_bytes, "log cleaner compression state changed"); + // A real run keeps the archive on disk, so freed = input - archive. In + // dry-run we cannot know the true archive size, so estimate it with a + // deliberately conservative ratio: this keeps the projected freed bytes + // from overstating what a real run would reclaim (previously output_bytes + // was 0, so dry-run reported the full input as freed). + let input_bytes = std::fs::metadata(path).map(|m| m.len()).unwrap_or(0); + let estimated_output_bytes = (input_bytes as f64 * DRY_RUN_ESTIMATED_ARCHIVE_RATIO) as u64; + info!(event = EVENT_LOG_CLEANER_COMPRESSION_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_LOG_CLEANER, state = "dry_run_compress", file = ?path, archive = ?archive_path, input_bytes, estimated_output_bytes, "log cleaner compression state changed"); return Ok(CompressionOutput { archive_path: archive_path.to_path_buf(), algorithm_used, input_bytes, - output_bytes: 0, + output_bytes: estimated_output_bytes, }); } @@ -161,37 +283,74 @@ where // helper remains side-effect free when the caller is evaluating policy. let input = File::open(path)?; + // Read the source mode from the already-open fd (no extra path stat, no + // TOCTOU) so the temp file is created restrictive — never wider than the + // source, default 0600 — instead of the world-readable 0644 that + // File::create would leave until a post-write chmod. + #[cfg(unix)] + let source_mode = { + use std::os::unix::fs::PermissionsExt; + input.metadata().ok().map(|m| m.permissions().mode()) + }; + #[cfg(not(unix))] + let source_mode: Option = None; + // Write to a temporary file first and then atomically rename it into place // so callers never observe a partially written archive at `archive_path`. let mut tmp_name = archive_path.as_os_str().to_owned(); tmp_name.push(".tmp"); let tmp_archive_path = std::path::PathBuf::from(tmp_name); - let output = File::create(&tmp_archive_path)?; + let output = create_tmp_archive(&tmp_archive_path, source_mode)?; + // Arm a guard so any early return *or panic* between here and the final + // rename removes the incomplete temporary archive instead of leaking it. + let mut tmp_guard = TmpFileGuard::new(tmp_archive_path.clone()); let mut reader = BufReader::new(input); let writer = BufWriter::new(output); - if let Err(e) = writer_fn(&mut reader, writer) { - // Best-effort cleanup of the incomplete temporary archive. - let _ = std::fs::remove_file(&tmp_archive_path); - return Err(e); - } + let (written, out_file) = writer_fn(&mut reader, writer)?; - // Preserve Unix mode bits so rotated archives keep the same access policy. + // Durability: force the archive contents to stable storage *before* the + // rename. Without this, a crash after rename but before the page cache is + // flushed can leave a zero-length/truncated archive while the source is + // later deleted — permanent log/audit data loss. Renaming to a brand-new + // name (the `exists()` guard above guarantees this) means ext4's + // `auto_da_alloc` safety net does not apply, so the fsync is mandatory. + out_file.sync_all()?; + drop(out_file); + + // Reapply the exact source mode: create_tmp_archive already set it at + // creation, but umask may have narrowed it, so restore it precisely — + // reusing the value already read from the fd rather than stat-ing again. #[cfg(unix)] - { + if let Some(mode) = source_mode { use std::os::unix::fs::PermissionsExt; - - if let Ok(src_meta) = std::fs::metadata(path) { - let mode = src_meta.permissions().mode(); - let _ = std::fs::set_permissions(&tmp_archive_path, std::fs::Permissions::from_mode(mode)); - } + let _ = std::fs::set_permissions(&tmp_archive_path, std::fs::Permissions::from_mode(mode)); } // Atomically move the fully written temp file into its final location. std::fs::rename(&tmp_archive_path, archive_path)?; + // The archive now lives at its final path; disarm the guard so it is not + // removed on the way out. + tmp_guard.disarm(); - let output_bytes = std::fs::metadata(archive_path).map(|m| m.len()).unwrap_or(0); + // Persist the new directory entry so the rename survives a crash. Failure + // here is best-effort (e.g. platforms that reject opening a directory). + sync_parent_dir(archive_path); + + // Use the copied byte count as the authoritative input size rather than a + // second metadata read on the source (which could silently read 0 on error + // and skew freed-byte accounting). + let input_bytes = written; + let output_bytes = match std::fs::metadata(archive_path) { + Ok(meta) => meta.len(), + Err(err) => { + // Conservative: assume no savings instead of silently reporting 0, + // which would overstate freed bytes as the full input. + warn!(event = EVENT_LOG_CLEANER_COMPRESSION_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_LOG_CLEANER, archive = ?archive_path, error = %err, result = "archive_metadata_read_failed", "log cleaner compression state changed"); + input_bytes + } + }; debug!( event = EVENT_LOG_CLEANER_COMPRESSION_STATE, component = LOG_COMPONENT_OBS, diff --git a/crates/obs/src/cleaner/core.rs b/crates/obs/src/cleaner/core.rs index ae9d906eb..3f9188cf6 100644 --- a/crates/obs/src/cleaner/core.rs +++ b/crates/obs/src/cleaner/core.rs @@ -28,7 +28,7 @@ use crate::global::{ }; use crossbeam_channel::bounded; use crossbeam_deque::{Injector, Steal, Stealer, Worker}; -use crossbeam_utils::thread; +use crossbeam_utils::{Backoff, thread}; use metrics::{counter, gauge, histogram}; use rustfs_config::DEFAULT_LOG_KEEP_FILES; use std::path::PathBuf; @@ -41,6 +41,11 @@ const LOG_COMPONENT_OBS: &str = "obs"; const LOG_SUBSYSTEM_LOG_CLEANER: &str = "log_cleaner"; const EVENT_LOG_CLEANER_STATE: &str = "log_cleaner_state"; const SECONDS_PER_DAY: u64 = 24 * 60 * 60; +/// Absolute ceiling on parallel compression workers, independent of config, so +/// a misconfigured `parallel_workers` on a directory with thousands of rotated logs +/// cannot spawn thousands of threads (each of which may also start codec +/// threads). +const MAX_PARALLEL_COMPRESS_WORKERS: usize = 64; const MAX_RETENTION_DAYS_BEFORE_SATURATION: u64 = u64::MAX / SECONDS_PER_DAY; fn compressed_file_retention_window(days: u64) -> Duration { @@ -115,6 +120,16 @@ impl LogCleaner { LogCleanerBuilder::new(log_dir, file_pattern, active_filename) } + /// Effective gzip level after clamping — what compression actually uses. + pub fn effective_gzip_level(&self) -> u32 { + self.gzip_compression_level + } + + /// Effective zstd level after clamping — what compression actually uses. + pub fn effective_zstd_level(&self) -> i32 { + self.zstd_compression_level + } + /// Perform one full cleanup pass. pub fn cleanup(&self) -> Result<(usize, u64), std::io::Error> { if !self.log_dir.exists() { @@ -175,10 +190,10 @@ impl LogCleaner { } } - if !compressed_archives.is_empty() && self.compressed_file_retention_days > 0 { - let expired = self.select_expired_compressed(&mut compressed_archives); - if !expired.is_empty() { - let (d, f) = self.delete_files(&expired)?; + if !compressed_archives.is_empty() { + let to_delete = self.select_archives_to_delete(&mut compressed_archives); + if !to_delete.is_empty() { + let (d, f) = self.delete_files(&to_delete)?; total_deleted += d; total_freed += f; } @@ -235,8 +250,16 @@ impl LogCleaner { to_delete } - /// Select compressed archives whose age exceeds the archive retention window. - fn select_expired_compressed(&self, files: &mut [FileInfo]) -> Vec { + /// Select compressed archives to delete: those older than the retention + /// window, plus — regardless of retention — the oldest archives needed to + /// bring the archive set back under `max_total_size_bytes`. + /// + /// The byte cap is the crucial part: with `compressed_file_retention_days` + /// set to 0 (retention disabled) and compression enabled, age-based expiry + /// never fires, so without this cap archives would grow without bound (the + /// cap on regular logs does not cover them). Bounding the archive set by the + /// same byte budget keeps disk usage finite even when retention is off. + fn select_archives_to_delete(&self, files: &mut [FileInfo]) -> Vec { if self.compressed_file_retention_days > MAX_RETENTION_DAYS_BEFORE_SATURATION { warn!( event = EVENT_LOG_CLEANER_STATE, @@ -248,19 +271,54 @@ impl LogCleaner { "log cleaner state changed" ); } - let retention = compressed_file_retention_window(self.compressed_file_retention_days); - let now = SystemTime::now(); - let mut expired = Vec::new(); - for file in files { - if let Ok(age) = now.duration_since(file.modified) - && age > retention - { - expired.push(file.clone()); + // Oldest first so both age expiry and the byte-cap trim evict the + // longest-retained archives before newer ones. + files.sort_by_key(|f| f.modified); + let now = SystemTime::now(); + let retention = compressed_file_retention_window(self.compressed_file_retention_days); + let retention_active = self.compressed_file_retention_days > 0; + + let mut selected = vec![false; files.len()]; + + // 1. Age-based expiry (only when retention is enabled). + if retention_active { + for (idx, file) in files.iter().enumerate() { + if let Ok(age) = now.duration_since(file.modified) + && age > retention + { + selected[idx] = true; + } } } - expired + // 2. Byte-cap trim: drop the oldest surviving archives until the set + // fits under the configured total. 0 means "no cap". + if self.max_total_size_bytes > 0 { + let mut remaining: u64 = files + .iter() + .enumerate() + .filter(|(idx, _)| !selected[*idx]) + .map(|(_, file)| file.size) + .sum(); + for (idx, file) in files.iter().enumerate() { + if remaining <= self.max_total_size_bytes { + break; + } + if selected[idx] { + continue; + } + selected[idx] = true; + remaining = remaining.saturating_sub(file.size); + } + } + + files + .iter() + .enumerate() + .filter(|(idx, _)| selected[*idx]) + .map(|(_, file)| file.clone()) + .collect() } /// Parallel compressor with work stealing. @@ -275,7 +333,7 @@ impl LogCleaner { return self.serial_compress_and_delete(files); } - let worker_count = self.parallel_workers.min(files.len()).max(1); + let worker_count = self.parallel_workers.min(files.len()).clamp(1, MAX_PARALLEL_COMPRESS_WORKERS); if worker_count <= 1 { return self.serial_compress_and_delete(files); } @@ -317,6 +375,11 @@ impl LogCleaner { .wrapping_mul(6364136223846793005) .wrapping_add(1442695040888963407); + // Escalating spin->yield backoff for the transient windows + // where work is being redistributed, instead of a hot + // `continue`/`yield_now` loop that burns CPU. + let backoff = Backoff::new(); + loop { // Search order: local FIFO -> global injector batch -> // random victim stealers. @@ -327,7 +390,10 @@ impl LogCleaner { Steal::Success(file) => { Some(file) } - Steal::Retry => continue, + Steal::Retry => { + backoff.snooze(); + continue; + } Steal::Empty => { let stolen = Self::steal_from_victims( worker_id, @@ -351,10 +417,13 @@ impl LogCleaner { }; let Some(file) = task else { - std::thread::yield_now(); + backoff.snooze(); continue; }; + // Got work: reset the backoff so the next idle spell + // starts from a cheap spin again. + backoff.reset(); let mut file = file; let compressed = match compress_file(&file.path, &options) { Ok(output) => { @@ -415,6 +484,11 @@ impl LogCleaner { let (deleted, freed) = self.delete_files(&deletable)?; let elapsed = started_at.elapsed().as_secs_f64(); + // NOTE on metric scope: attempts/successes count only *victim* steals + // (not the injector `steal_batch_and_pop`), and a multi-task stolen + // batch counts as a single success. So this rate reflects inter-worker + // rebalancing efficiency, not total task acquisition — read it as a + // relative health signal, not an absolute throughput measure. let attempts = steal_attempts.load(Ordering::Relaxed); let successes = steal_successes.load(Ordering::Relaxed); let success_rate = if attempts == 0 { @@ -603,6 +677,16 @@ impl LogCleaner { } } + // Durability: persist the directory entries so the unlinks cannot be + // reordered ahead of the archive data that justified them. Best-effort; + // opening a directory as a file is not portable (e.g. Windows). + if !self.dry_run + && deleted > 0 + && let Ok(dir) = std::fs::File::open(&self.log_dir) + { + let _ = dir.sync_all(); + } + Ok((deleted, freed)) } } @@ -774,11 +858,40 @@ impl LogCleanerBuilder { /// Invalid glob patterns are ignored rather than failing construction, and /// codec-related numeric values are clamped into safe ranges. pub fn build(self) -> LogCleaner { - let patterns = self - .exclude_patterns - .into_iter() - .filter_map(|p| glob::Pattern::new(&p).ok()) - .collect(); + // Defense-in-depth for the active-file guard: the scanner protects the + // live log by exact filename equality against `active_filename`. An + // empty value silently disables that protection, so a non-empty + // `file_pattern` could then make the live log a deletion candidate. + // Warn instead of failing so misuse of the public builder is visible. + if self.active_filename.trim().is_empty() && !self.file_pattern.trim().is_empty() { + warn!( + event = EVENT_LOG_CLEANER_STATE, + component = LOG_COMPONENT_OBS, + subsystem = LOG_SUBSYSTEM_LOG_CLEANER, + result = "empty_active_filename", + "active log filename is empty; the active-file exclusion is disabled and the live log may become a deletion candidate" + ); + } + + // Surface, rather than silently drop, invalid exclude globs: a dropped + // pattern turns "protect this file" into "delete this file", so an + // operator's typo (or a literal comma splitting a char-class) must be + // visible instead of failing open. + let mut patterns = Vec::new(); + for raw in self.exclude_patterns { + match glob::Pattern::new(&raw) { + Ok(pattern) => patterns.push(pattern), + Err(err) => warn!( + event = EVENT_LOG_CLEANER_STATE, + component = LOG_COMPONENT_OBS, + subsystem = LOG_SUBSYSTEM_LOG_CLEANER, + result = "invalid_exclude_pattern", + pattern = %raw, + error = %err, + "log cleaner state changed" + ), + } + } LogCleaner { log_dir: self.log_dir, @@ -789,7 +902,9 @@ impl LogCleanerBuilder { max_total_size_bytes: self.max_total_size_bytes, max_single_file_size_bytes: self.max_single_file_size_bytes, compress_old_files: self.compress_old_files, - gzip_compression_level: self.gzip_compression_level.clamp(1, 9), + // gzip level 0 is a valid "store" (no compression) mode, so keep the + // lower bound at 0 rather than silently bumping it to 1. + gzip_compression_level: self.gzip_compression_level.clamp(0, 9), compressed_file_retention_days: self.compressed_file_retention_days, exclude_patterns: patterns, delete_empty_files: self.delete_empty_files, @@ -798,7 +913,9 @@ impl LogCleanerBuilder { compression_algorithm: self.compression_algorithm, parallel_compress: self.parallel_compress, parallel_workers: self.parallel_workers.max(1), - zstd_compression_level: self.zstd_compression_level.clamp(1, 21), + // zstd level 0 means "codec default"; the real maximum is 22, not + // 21, so preserve both boundaries instead of silently narrowing. + zstd_compression_level: self.zstd_compression_level.clamp(0, 22), zstd_fallback_to_gzip: self.zstd_fallback_to_gzip, zstd_workers: self.zstd_workers.max(1), } diff --git a/crates/obs/src/cleaner/mod.rs b/crates/obs/src/cleaner/mod.rs index 2fe7465f1..b65be9456 100644 --- a/crates/obs/src/cleaner/mod.rs +++ b/crates/obs/src/cleaner/mod.rs @@ -303,4 +303,279 @@ mod tests { fn test_parallel_cleanup_handles_results_within_channel_capacity() -> std::io::Result<()> { assert_parallel_cleanup_completes(6, 4) } + + // ---- OLC-14: safety/correctness regression coverage ---- + + use std::time::SystemTime; + + fn set_mtime_ago(path: &Path, ago: Duration) { + let when = SystemTime::now() - ago; + File::options() + .write(true) + .open(path) + .expect("open for mtime") + .set_modified(when) + .expect("set mtime"); + } + + #[cfg(unix)] + #[test] + fn symlink_matching_pattern_is_never_deleted() -> std::io::Result<()> { + let tmp = TempDir::new()?; + let dir = tmp.path().to_path_buf(); + let outside = TempDir::new()?; + let target = outside.path().join("precious.txt"); + std::fs::write(&target, b"precious")?; + // A symlink named to match the managed pattern. + std::os::unix::fs::symlink(&target, dir.join("app.log.2024-01-01"))?; + create_log_file(&dir, "app.log.2024-01-02", 1024)?; + + // keep_files=0 would delete every managed *regular* file. + let cleaner = make_cleaner(dir.clone(), 0, 0); + let _ = cleaner.cleanup()?; + + assert!(target.exists(), "external symlink target must never be deleted"); + assert!( + dir.join("app.log.2024-01-01").exists(), + "symlink entry must survive (skipped via symlink_metadata, not followed)" + ); + Ok(()) + } + + #[test] + fn expired_archives_deleted_fresh_kept() -> std::io::Result<()> { + let tmp = TempDir::new()?; + let dir = tmp.path().to_path_buf(); + create_log_file(&dir, "app.log.2024-01-01.gz", 100)?; + create_log_file(&dir, "app.log.2024-01-02.gz", 100)?; + set_mtime_ago(&dir.join("app.log.2024-01-01.gz"), Duration::from_secs(2 * 24 * 3600)); + + let cleaner = LogCleaner::builder(dir.clone(), "app.log.".to_string(), "app.log".to_string()) + .match_mode(FileMatchMode::Prefix) + .keep_files(100) + .compressed_file_retention_days(1) + .min_file_age_seconds(0) + .build(); + let (deleted, _) = cleaner.cleanup()?; + + assert_eq!(deleted, 1, "only the aged archive should expire"); + assert!(!dir.join("app.log.2024-01-01.gz").exists()); + assert!(dir.join("app.log.2024-01-02.gz").exists()); + Ok(()) + } + + #[test] + fn archives_bounded_by_byte_cap_when_retention_disabled() -> std::io::Result<()> { + let tmp = TempDir::new()?; + let dir = tmp.path().to_path_buf(); + create_log_file(&dir, "app.log.2024-01-01.gz", 100)?; + create_log_file(&dir, "app.log.2024-01-02.gz", 100)?; + // Age the first so the oldest-first byte-cap trim evicts it. + set_mtime_ago(&dir.join("app.log.2024-01-01.gz"), Duration::from_secs(100)); + + let cleaner = LogCleaner::builder(dir.clone(), "app.log.".to_string(), "app.log".to_string()) + .match_mode(FileMatchMode::Prefix) + .keep_files(100) + .compressed_file_retention_days(0) // retention disabled + .max_total_size_bytes(150) // 200 total > 150 -> trim the oldest + .min_file_age_seconds(0) + .build(); + let (deleted, _) = cleaner.cleanup()?; + + assert_eq!(deleted, 1, "byte cap must bound archives even with retention off"); + assert!(!dir.join("app.log.2024-01-01.gz").exists(), "oldest archive trimmed"); + assert!(dir.join("app.log.2024-01-02.gz").exists()); + Ok(()) + } + + #[test] + fn gz_and_zst_files_classified_as_archives() -> std::io::Result<()> { + let tmp = TempDir::new()?; + let dir = tmp.path().to_path_buf(); + create_log_file(&dir, "app.log.2024-01-01.gz", 100)?; + create_log_file(&dir, "app.log.2024-01-02.zst", 100)?; + create_log_file(&dir, "app.log.2024-01-03", 100)?; // plain log + + let result = scanner::scan_log_directory(&dir, "app.log.", Some("app.log"), FileMatchMode::Prefix, &[], 0, false, false)?; + assert_eq!(result.compressed_archives.len(), 2, "gz and zst are archives"); + assert_eq!(result.logs.len(), 1, "only the plain file is a regular log"); + Ok(()) + } + + #[test] + fn max_single_file_size_deletes_only_oversized() -> std::io::Result<()> { + let tmp = TempDir::new()?; + let dir = tmp.path().to_path_buf(); + create_log_file(&dir, "app.log.2024-01-01", 100)?; + create_log_file(&dir, "app.log.2024-01-02", 100)?; + create_log_file(&dir, "app.log.2024-01-03", 5000)?; + + let cleaner = LogCleaner::builder(dir.clone(), "app.log.".to_string(), "app.log".to_string()) + .match_mode(FileMatchMode::Prefix) + .keep_files(3) + .max_total_size_bytes(0) + .max_single_file_size_bytes(1000) + .min_file_age_seconds(0) + .build(); + let (deleted, _) = cleaner.cleanup()?; + + assert_eq!(deleted, 1); + assert!(!dir.join("app.log.2024-01-03").exists(), "oversized file removed"); + assert!(dir.join("app.log.2024-01-01").exists()); + assert!(dir.join("app.log.2024-01-02").exists()); + Ok(()) + } + + #[test] + fn min_age_protects_fresh_nonempty_log() -> std::io::Result<()> { + let tmp = TempDir::new()?; + let dir = tmp.path().to_path_buf(); + create_log_file(&dir, "app.log.2024-01-01", 1024)?; + + let cleaner = LogCleaner::builder(dir.clone(), "app.log.".to_string(), "app.log".to_string()) + .match_mode(FileMatchMode::Prefix) + .keep_files(0) + .min_file_age_seconds(3600) + .build(); + let (deleted, _) = cleaner.cleanup()?; + + assert_eq!(deleted, 0, "fresh non-empty log younger than min age must be untouched"); + assert!(dir.join("app.log.2024-01-01").exists()); + Ok(()) + } + + #[test] + fn active_file_matching_pattern_is_protected() -> std::io::Result<()> { + let tmp = TempDir::new()?; + let dir = tmp.path().to_path_buf(); + // The active file both equals active_filename AND matches the suffix pattern. + create_log_file(&dir, "current.rustfs.log", 1024)?; + create_log_file(&dir, "old.rustfs.log", 1024)?; + + let cleaner = LogCleaner::builder(dir.clone(), ".rustfs.log".to_string(), "current.rustfs.log".to_string()) + .match_mode(FileMatchMode::Suffix) + .keep_files(0) + .min_file_age_seconds(0) + .build(); + let _ = cleaner.cleanup()?; + + assert!(dir.join("current.rustfs.log").exists(), "active file must never be deleted"); + assert!(!dir.join("old.rustfs.log").exists(), "non-active rotated log removed"); + Ok(()) + } + + #[test] + fn invalid_exclude_glob_does_not_abort_build() -> std::io::Result<()> { + let tmp = TempDir::new()?; + let dir = tmp.path().to_path_buf(); + create_log_file(&dir, "app.log.keep", 100)?; + create_log_file(&dir, "app.log.drop", 100)?; + + // "[invalid" is dropped with a warning; "*keep*" is a valid exclude. + let cleaner = LogCleaner::builder(dir.clone(), "app.log.".to_string(), "app.log".to_string()) + .match_mode(FileMatchMode::Prefix) + .keep_files(0) + .exclude_patterns(vec!["[invalid".to_string(), "*keep*".to_string()]) + .min_file_age_seconds(0) + .build(); + let _ = cleaner.cleanup()?; + + assert!(dir.join("app.log.keep").exists(), "valid exclude must still protect its file"); + assert!(!dir.join("app.log.drop").exists(), "non-excluded rotated log removed"); + Ok(()) + } + + #[test] + fn dry_run_with_compression_creates_no_archive() -> std::io::Result<()> { + let tmp = TempDir::new()?; + let dir = tmp.path().to_path_buf(); + create_log_file(&dir, "app.log.2024-01-01", 1024)?; + + let cleaner = LogCleaner::builder(dir.clone(), "app.log.".to_string(), "app.log".to_string()) + .match_mode(FileMatchMode::Prefix) + .keep_files(0) + .compress_old_files(true) + .compression_algorithm(CompressionAlgorithm::Gzip) + .parallel_compress(false) + .min_file_age_seconds(0) + .dry_run(true) + .build(); + let (deleted, freed) = cleaner.cleanup()?; + + assert!(deleted > 0, "dry-run reports intended deletions"); + assert!(freed > 0, "dry-run reports projected freed bytes"); + assert!(dir.join("app.log.2024-01-01").exists(), "dry-run must not delete"); + assert!(!dir.join("app.log.2024-01-01.gz").exists(), "dry-run must not create an archive"); + Ok(()) + } + + #[test] + fn gzip_archive_round_trips_and_source_removed() -> std::io::Result<()> { + use std::io::Read; + let tmp = TempDir::new()?; + let dir = tmp.path().to_path_buf(); + let content = vec![b'A'; 4096]; + { + let mut f = File::create(dir.join("app.log.2024-01-01"))?; + f.write_all(&content)?; + f.flush()?; + } + + let cleaner = LogCleaner::builder(dir.clone(), "app.log.".to_string(), "app.log".to_string()) + .match_mode(FileMatchMode::Prefix) + .keep_files(0) + .compress_old_files(true) + .compression_algorithm(CompressionAlgorithm::Gzip) + .parallel_compress(false) + .min_file_age_seconds(0) + .build(); + let _ = cleaner.cleanup()?; + + assert!(!dir.join("app.log.2024-01-01").exists(), "source removed after compression"); + let gz = dir.join("app.log.2024-01-01.gz"); + assert!(gz.exists(), "archive created"); + let mut decoded = Vec::new(); + flate2::read::GzDecoder::new(File::open(&gz)?).read_to_end(&mut decoded)?; + assert_eq!(decoded, content, "archive must decompress to the original bytes"); + Ok(()) + } + + #[test] + fn idempotent_archive_branch_trusts_valid_archive() -> std::io::Result<()> { + let tmp = TempDir::new()?; + let dir = tmp.path().to_path_buf(); + let content = vec![b'B'; 2048]; + let write_source = || -> std::io::Result<()> { + let mut f = File::create(dir.join("app.log.2024-01-01"))?; + f.write_all(&content)?; + f.flush() + }; + write_source()?; + + let build = || { + LogCleaner::builder(dir.clone(), "app.log.".to_string(), "app.log".to_string()) + .match_mode(FileMatchMode::Prefix) + .keep_files(0) + .compress_old_files(true) + .compression_algorithm(CompressionAlgorithm::Gzip) + .parallel_compress(false) + .min_file_age_seconds(0) + .build() + }; + + // Pass 1: create the archive and delete the source. + let _ = build().cleanup()?; + let gz = dir.join("app.log.2024-01-01.gz"); + assert!(gz.exists()); + let archive_len = std::fs::metadata(&gz)?.len(); + + // Recreate the source; pass 2 must trust the existing valid archive and + // delete the source again without rewriting the archive. + write_source()?; + let (deleted, _) = build().cleanup()?; + assert_eq!(deleted, 1); + assert!(!dir.join("app.log.2024-01-01").exists(), "source deleted via idempotent branch"); + assert_eq!(std::fs::metadata(&gz)?.len(), archive_len, "existing valid archive unchanged"); + Ok(()) + } } diff --git a/crates/obs/src/cleaner/scanner.rs b/crates/obs/src/cleaner/scanner.rs index c8c371ade..e189a3ddb 100644 --- a/crates/obs/src/cleaner/scanner.rs +++ b/crates/obs/src/cleaner/scanner.rs @@ -34,6 +34,11 @@ use tracing::debug; const LOG_COMPONENT_OBS: &str = "obs"; const LOG_SUBSYSTEM_LOG_CLEANER: &str = "log_cleaner"; const EVENT_LOG_CLEANER_SCAN_STATE: &str = "log_cleaner_scan_state"; +/// Small grace window before removing an orphan `*.tmp` archive. Orphans are +/// never live-written after the rename that would have promoted them, so they +/// are exempt from `min_file_age_seconds`; the grace only avoids racing a temp +/// file another process may be mid-write on a shared directory. +const ORPHAN_TMP_GRACE_SECONDS: u64 = 60; /// Result of a single pass directory scan. /// @@ -146,9 +151,9 @@ pub(super) fn scan_log_directory( let matched_suffix = CompressionAlgorithm::compressed_suffixes() .into_iter() .find(|suffix| filename.ends_with(suffix)); - let matched_tmp_suffix = compressed_tmp_suffix(filename); + let matched_tmp_suffix_len = compressed_tmp_suffix_len(filename); let is_compressed = matched_suffix.is_some(); - let is_tmp_archive = matched_tmp_suffix.is_some(); + let is_tmp_archive = matched_tmp_suffix_len.is_some(); // Perform matching on the logical log filename. // Examples: @@ -159,8 +164,8 @@ pub(super) fn scan_log_directory( // to both raw and already-compressed generations. let name_to_match = if let Some(suffix) = matched_suffix { &filename[..filename.len() - suffix.len()] - } else if let Some(suffix) = matched_tmp_suffix { - &filename[..filename.len() - suffix.len()] + } else if let Some(len) = matched_tmp_suffix_len { + &filename[..filename.len() - len] } else { filename }; @@ -184,7 +189,9 @@ pub(super) fn scan_log_directory( let age = now.duration_since(modified).ok(); if is_tmp_archive { - if !is_old_enough_for_cleanup(age, min_file_age_seconds) { + // Orphan temp archives are exempt from min_file_age_seconds (they + // are not live logs); only a short grace window applies. + if !is_old_enough_for_cleanup(age, ORPHAN_TMP_GRACE_SECONDS) { continue; } @@ -253,8 +260,14 @@ pub(super) fn is_excluded(filename: &str, patterns: &[glob::Pattern]) -> bool { patterns.iter().any(|p| p.matches(filename)) } -fn compressed_tmp_suffix(filename: &str) -> Option<&'static str> { - [".gz.tmp", ".zst.tmp"].into_iter().find(|suffix| filename.ends_with(suffix)) +/// Length of the matched compressed-temp suffix (e.g. `.gz.tmp`), derived from +/// [`CompressionAlgorithm::compressed_suffixes`] rather than hardcoded, so a new +/// codec cannot leave orphan `*..tmp` files the scanner never recognizes. +fn compressed_tmp_suffix_len(filename: &str) -> Option { + CompressionAlgorithm::compressed_suffixes().into_iter().find_map(|suffix| { + let tmp_suffix = format!("{suffix}.tmp"); + filename.ends_with(&tmp_suffix).then_some(tmp_suffix.len()) + }) } fn is_old_enough_for_cleanup(age: Option, min_file_age_seconds: u64) -> bool { @@ -304,6 +317,9 @@ mod tests { let dir = tmp.path(); let filename = "2026-07-08.rustfs.log.gz.tmp"; create_empty_file(dir, filename)?; + // Age the orphan past the grace window so it is eligible for removal. + let old = SystemTime::now() - Duration::from_secs(ORPHAN_TMP_GRACE_SECONDS + 60); + File::options().write(true).open(dir.join(filename))?.set_modified(old)?; let _ = scan_log_directory(dir, ".rustfs.log", Some("active.log"), FileMatchMode::Suffix, &[], 0, true, false)?; @@ -311,6 +327,22 @@ mod tests { Ok(()) } + #[test] + fn fresh_orphan_tmp_kept_within_grace() -> std::io::Result<()> { + let tmp = TempDir::new()?; + let dir = tmp.path(); + // `.zst.tmp` also verifies the suffix is derived from compressed_suffixes. + let filename = "2026-07-08.rustfs.log.zst.tmp"; + create_empty_file(dir, filename)?; + + // Fresh (age ~0) is within the grace window; min_file_age is irrelevant + // to orphan temp files and must not keep them around beyond the grace. + let _ = scan_log_directory(dir, ".rustfs.log", Some("active.log"), FileMatchMode::Suffix, &[], 999_999, true, false)?; + + assert!(dir.join(filename).exists(), "fresh orphan tmp should be kept within grace"); + Ok(()) + } + #[test] fn cleanup_age_gate_requires_known_old_enough_age() { assert!(!is_old_enough_for_cleanup(Some(Duration::from_secs(0)), 1)); diff --git a/crates/obs/src/cleaner/types.rs b/crates/obs/src/cleaner/types.rs index 201589aaa..7cfc97a92 100644 --- a/crates/obs/src/cleaner/types.rs +++ b/crates/obs/src/cleaner/types.rs @@ -28,6 +28,11 @@ use rustfs_config::observability::{ use std::fmt; use std::path::PathBuf; use std::time::SystemTime; +use tracing::warn; + +const LOG_COMPONENT_OBS: &str = "obs"; +const LOG_SUBSYSTEM_LOG_CLEANER: &str = "log_cleaner"; +const EVENT_LOG_CLEANER_STATE: &str = "log_cleaner_state"; /// Strategy for matching log files against a pattern. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -55,9 +60,25 @@ impl FileMatchMode { /// configuration handling permissive and aligned with the historical /// cleaner default used by rolling log filenames. pub fn from_config_str(value: &str) -> Self { - if value.trim().eq_ignore_ascii_case(DEFAULT_OBS_LOG_MATCH_MODE_PREFIX) { + let trimmed = value.trim(); + if trimmed.eq_ignore_ascii_case(DEFAULT_OBS_LOG_MATCH_MODE_PREFIX) { Self::Prefix + } else if trimmed.eq_ignore_ascii_case(DEFAULT_OBS_LOG_MATCH_MODE) { + Self::Suffix } else { + // A non-empty unrecognized value is almost certainly a typo (e.g. + // "prefixx"); warn so the silent fallback to suffix is visible. + if !trimmed.is_empty() { + warn!( + event = EVENT_LOG_CLEANER_STATE, + component = LOG_COMPONENT_OBS, + subsystem = LOG_SUBSYSTEM_LOG_CLEANER, + result = "unknown_match_mode", + value = %value, + fallback = DEFAULT_OBS_LOG_MATCH_MODE, + "log cleaner state changed" + ); + } Self::Suffix } } @@ -97,7 +118,26 @@ impl CompressionAlgorithm { /// to the crate default so observability startup remains resilient. pub fn from_config_str(value: &str) -> Self { let normalized = value.trim().to_ascii_lowercase(); - Self::parse_normalized(&normalized).unwrap_or_default() + match Self::parse_normalized(&normalized) { + Some(algorithm) => algorithm, + None => { + let fallback = Self::default(); + // A non-empty unrecognized value is almost certainly a typo; + // warn so the silent fallback to the default codec is visible. + if !normalized.is_empty() { + warn!( + event = EVENT_LOG_CLEANER_STATE, + component = LOG_COMPONENT_OBS, + subsystem = LOG_SUBSYSTEM_LOG_CLEANER, + result = "unknown_compression_algorithm", + value = %value, + fallback = %fallback, + "log cleaner state changed" + ); + } + fallback + } + } } /// Archive suffix (without dot) used for this algorithm. @@ -154,13 +194,12 @@ impl fmt::Display for CompressionAlgorithm { /// Worker-thread default used by parallel compressor. /// -/// The worker count follows CPU capacity but stays within [4, 8] to keep -/// throughput stable and avoid oversubscription. The lower bound helps the -/// work-stealing path on small machines still exercise concurrency, while the -/// upper bound avoids swamping the host when each task may also use internal -/// codec threads. +/// The worker count follows CPU capacity, capped at 8 to avoid swamping the +/// host when each task may also use internal codec threads. The lower bound is +/// 1 (not 4) so a 1-2 vCPU container is not forced to run several concurrent +/// compressions that monopolize its CPU. pub fn default_parallel_workers() -> usize { - num_cpus::get().clamp(4, 8) + num_cpus::get().clamp(1, 8) } /// Metadata for a single log file discovered by the scanner. diff --git a/crates/obs/src/telemetry/local.rs b/crates/obs/src/telemetry/local.rs index 6e6f50a46..b0a9d947b 100644 --- a/crates/obs/src/telemetry/local.rs +++ b/crates/obs/src/telemetry/local.rs @@ -654,15 +654,32 @@ pub fn spawn_cleanup_task( component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_LOCAL_LOGGING, state = "configured", + match_mode = %match_mode, compression_algorithm = %compression_algorithm, parallel_compress, parallel_workers, - zstd_level, + // Echo the *effective* (post-clamp) levels so the log matches what the + // codec actually runs with, not the raw configured value. + gzip_level = cleaner.effective_gzip_level(), + zstd_level = cleaner.effective_zstd_level(), zstd_fallback_to_gzip, zstd_workers, "log cleaner state" ); + // retention_days == 0 means "keep archives forever" (age expiry disabled), + // which is easy to misread as "delete immediately". Warn so an operator who + // enabled compression is aware archives rely solely on the byte cap. + if compress && retention_days == 0 { + warn!( + event = EVENT_LOG_CLEANER_STATE, + component = LOG_COMPONENT_OBS, + subsystem = LOG_SUBSYSTEM_LOCAL_LOGGING, + result = "archive_retention_disabled", + "compressed archives are kept forever (retention_days=0); disk is bounded only by max_total_size_bytes" + ); + } + tokio::spawn(async move { let mut interval = tokio::time::interval(Duration::from_secs(cleanup_interval)); loop {