mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 08:18:18 +00:00
fix(obs/cleaner): harden log cleaner durability, symlink safety, and retention (audit OLC-01..14) (#4776)
* fix(obs/cleaner): fsync archive and log dir before deleting source logs OLC-01: the compression path flushed the BufWriter but never synced the archive data or the parent directory before renaming, and the source was then unlinked with no durability barrier. A crash after rename but before the page cache reached disk could leave a truncated/zero-length archive while the source was already gone — permanent log/audit data loss. Because the archive is always renamed to a brand-new name (guaranteed by the existing exists() guard), ext4 auto_da_alloc does not mask this. Hand the underlying File back from the writer closure, sync_all() it before rename, fsync the parent directory, and fsync the log directory after the unlinks so a delete cannot be reordered ahead of the archive it justified. Guard the temp file with an RAII cleanup so an early return or panic cannot leak a *.tmp orphan. Ref: rustfs/backlog#1194 (audit rustfs/backlog#1193) Co-Authored-By: heihutu <heihutu@gmail.com> * fix(obs/cleaner): create compression temp file with O_EXCL and O_NOFOLLOW OLC-03: the temp archive was opened with File::create at a predictable `<source>.gz.tmp` path with no O_EXCL/O_NOFOLLOW, so an actor with write access to the log directory could pre-plant that path as a symlink and have the compressor follow it — truncating and overwriting an arbitrary external file, then chmod-ing it to the source log's mode. This mirrors the symlink refusal already enforced on the deletion path (secure_delete). Route temp creation through create_tmp_archive(), which uses create_new (O_CREAT|O_EXCL) to refuse a pre-existing entry and, on Unix, O_NOFOLLOW to refuse a symlink at the final path component. Ref: rustfs/backlog#1196 (audit rustfs/backlog#1193) Co-Authored-By: heihutu <heihutu@gmail.com> * fix(obs/cleaner): create compression temp file with restrictive mode OLC-06: File::create left the temp archive world-readable (0644 & ~umask) for the entire duration of compressing a large log, exposing the full plaintext of a possibly-0600 audit log on shared hosts until the mode was copied only after the write completed. Pass the source mode into create_tmp_archive and open the temp file with it (default 0600) so it is restrictive from creation; the post-write chmod still tightens/matches the source mode exactly. Ref: rustfs/backlog#1199 (audit rustfs/backlog#1193) Co-Authored-By: heihutu <heihutu@gmail.com> * fix(obs/cleaner): validate existing archive before skipping recompression OLC-02: the idempotency guard used Path::exists() (which follows symlinks) and trusted whatever it found, then the caller deleted the source. A planted `<archive>.gz` symlink, or a zero-length/truncated archive left by a crashed run (OLC-01), would green-light deleting the source with no valid backup — data loss / log destruction. Replace exists() with symlink_metadata (no follow) and only treat the entry as a completed prior result when it is a regular, non-empty file whose header matches the codec magic (gzip 1f 8b / zstd 28 b5 2f fd). Anything else falls through to recompression, whose atomic create_new+rename replaces the bad entry (a symlink is replaced, never followed or deleted through). Ref: rustfs/backlog#1195 (audit rustfs/backlog#1193) Co-Authored-By: heihutu <heihutu@gmail.com> * fix(obs/cleaner): stop dry-run overstating reclaimed bytes for compression OLC-08: in dry-run, compress_with_writer returned output_bytes = 0, so projected_freed_bytes = input and delete_files reported the full input as freed. A real run keeps the archive on disk (freed = input - archive), so dry-run overstated reclaim by the whole archive footprint. Estimate the archive with a deliberately conservative ratio so the projection never exceeds what a real run reclaims. Ref: rustfs/backlog#1201 (audit rustfs/backlog#1193) Co-Authored-By: heihutu <heihutu@gmail.com> * fix(obs/cleaner): make freed-byte accounting resilient; document steal metric OLC-12: input/output byte sizes were read via metadata().unwrap_or(0), which silently reports 0 on failure and skews freed-byte metrics (input - 0 = full input, overstating reclaim). Use the copy() byte count as the authoritative input size and, when the archive metadata read fails, conservatively assume no savings instead of 0. Also document that the steal_success_rate counts only victim steals (batch = one success), so it reads as a relative rebalancing signal, not absolute task acquisition. Ref: rustfs/backlog#1205 (audit rustfs/backlog#1193) Co-Authored-By: heihutu <heihutu@gmail.com> * fix(obs/cleaner): preserve level-0 semantics, allow zstd 22, log effective levels OLC-09: build() and the codec calls clamped gzip/zstd levels to [1,9]/[1,21], silently rewriting gzip level 0 (store) and zstd level 0 (codec default) to 1 and blocking the legal zstd maximum of 22. Clamp to [0,9]/[0,22] so those meanings survive, and echo the effective (post-clamp) levels in the startup log via new effective_gzip_level()/effective_zstd_level() getters so the log matches what actually runs. Ref: rustfs/backlog#1202 (audit rustfs/backlog#1193) Co-Authored-By: heihutu <heihutu@gmail.com> * fix(obs/cleaner): bound compressed archives by byte cap; warn on retention=0 OLC-04: archive expiry was gated on compressed_file_retention_days > 0, so retention=0 disabled it entirely while compression kept producing archives, and max_total_size_bytes only ever bounded uncompressed logs — unbounded disk growth. Replace select_expired_compressed with select_archives_to_delete, which applies age expiry (when retention is on) and, regardless of retention, trims the oldest archives until the set fits under max_total_size_bytes. Also warn at startup when compression is on with retention=0 so the "keep forever" semantics are not mistaken for "delete immediately". Ref: rustfs/backlog#1197 (audit rustfs/backlog#1193) Co-Authored-By: heihutu <heihutu@gmail.com> * fix(obs/cleaner): warn on invalid exclude glob instead of dropping silently OLC-05: build() dropped unparseable exclude globs via filter_map(...ok()), so a typo (or a literal comma splitting a char-class in the config string) turned "protect this file" into "delete this file" with no signal. Log a warning per rejected pattern with the raw string and parse error so the misconfiguration is visible. Ref: rustfs/backlog#1198 (audit rustfs/backlog#1193) Co-Authored-By: heihutu <heihutu@gmail.com> * perf(obs/cleaner): backoff idle workers, cap worker count, lower small-host floor OLC-11: the work-stealing loop re-spun on Steal::Retry with no yield and used yield_now on the empty path, burning CPU during redistribution windows; worker_count had no upper bound so a mis-set parallel_workers over a directory of thousands of logs could spawn thousands of threads; and default_parallel_workers forced >=4 workers even on 1-2 vCPU hosts. Use crossbeam_utils::Backoff (spin->yield, reset on work) on the idle paths, clamp worker_count to MAX_PARALLEL_COMPRESS_WORKERS, and lower the default floor to 1. Ref: rustfs/backlog#1204 (audit rustfs/backlog#1193) Co-Authored-By: heihutu <heihutu@gmail.com> * fix(obs/cleaner): warn when active-file guard is disabled by empty filename OLC-13 (defense-in-depth): the scanner protects the live log purely by exact filename equality against active_filename. An empty active_filename silently disables that protection, so a non-empty file_pattern could make the live log a deletion candidate via the public builder. Warn in build() when that unsafe combination is configured. The audit's "never delete the newest match" structural guard is intentionally not implemented: it would conflict with the legitimate keep_files=0 semantics (purge all rotated logs). The naming contract is instead locked by regression tests (OLC-14). Ref: rustfs/backlog#1206 (audit rustfs/backlog#1193) Co-Authored-By: heihutu <heihutu@gmail.com> * fix(obs/cleaner): warn on unknown algorithm/match_mode, echo match_mode OLC-07: from_config_str silently fell back to defaults for unrecognized compression algorithm and match mode (any non-"prefix" value became Suffix), hiding operator typos like "prefixx" that could make the cleaner match no rotated logs. Warn on a non-empty unrecognized value in both parsers, and echo the resolved match_mode in the startup log alongside the algorithm. Ref: rustfs/backlog#1200 (audit rustfs/backlog#1193) Co-Authored-By: heihutu <heihutu@gmail.com> * fix(obs/cleaner): derive orphan .tmp suffixes and exempt them from min age OLC-10: orphan `*.gz.tmp`/`*.zst.tmp` cleanup was gated by min_file_age_seconds (default 3600), so crash-left orphans lingered up to an hour, and the tmp suffix list was hardcoded rather than derived from compressed_suffixes() — a new codec would leave `*.<ext>.tmp` orphans the scanner never recognizes. Derive the temp suffix from CompressionAlgorithm::compressed_suffixes(), and gate orphan removal on a small fixed grace window instead of min_file_age (orphans are never live-written after the rename that would promote them). Ref: rustfs/backlog#1203 (audit rustfs/backlog#1193) Co-Authored-By: heihutu <heihutu@gmail.com> * test(obs/cleaner): cover symlink, archive expiry, idempotency, and edge cases OLC-14: add regression tests for the previously-untested safety/correctness branches — symlink rejection (external target never deleted), archive age expiry vs fresh retention, archive byte-cap trim with retention disabled, gz/zst classification, max_single_file_size selection, min_age protecting a fresh non-empty log, active-file exclusion when the active name also matches the pattern, invalid exclude glob not aborting build, dry-run + compression creating no archive, gzip round-trip validity, and the idempotent-archive branch trusting a valid prior archive. Ref: rustfs/backlog#1207 (audit rustfs/backlog#1193) Co-Authored-By: heihutu <heihutu@gmail.com> * style(obs/cleaner): apply rustfmt and collapse nested if (clippy) Formatting-only cleanup over the audit fix series: rustfmt normalization of the multi-line expressions introduced in compress.rs/core.rs, plus collapsing the delete_files directory-fsync into a single let-chain to satisfy clippy::collapsible_if. No behavior change. Co-Authored-By: heihutu <heihutu@gmail.com> * refactor(obs/cleaner): collapse redundant source stats in compression path The per-issue fixes to compress_with_writer accumulated three metadata() syscalls on the source in the real compression path: an input_bytes read that was immediately shadowed by the copied byte count (a dead read), a source_mode read (OLC-06), and the pre-OLC-06 post-write chmod re-reading the same mode. Collapse to a single fd-based read — move the dry-run input_bytes read into the dry-run branch, read source_mode from the already-open fd (no path stat, no TOCTOU), and reuse it for the post-write chmod. Behavior is unchanged (same inode's mode, written for input size); 3 source stats -> 1. Co-Authored-By: heihutu <heihutu@gmail.com> * chore(obs/cleaner): fix typo flagged by CI (mis-set -> misconfigured) Co-Authored-By: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
Generated
+1
@@ -9593,6 +9593,7 @@ dependencies = [
|
||||
"futures-util",
|
||||
"glob",
|
||||
"jiff",
|
||||
"libc",
|
||||
"metrics",
|
||||
"num_cpus",
|
||||
"nvml-wrapper",
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -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<CompressionOutput, std::io::Error> {
|
||||
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<CompressionOu
|
||||
fn compress_zstd(path: &Path, level: i32, workers: usize, dry_run: bool) -> Result<CompressionOutput, std::io::Error> {
|
||||
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<PathBuf>,
|
||||
}
|
||||
|
||||
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
|
||||
/// `<archive>.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<u32>) -> std::io::Result<File> {
|
||||
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<F>(
|
||||
path: &Path,
|
||||
archive_path: &Path,
|
||||
@@ -130,30 +232,50 @@ fn compress_with_writer<F>(
|
||||
mut writer_fn: F,
|
||||
) -> Result<CompressionOutput, std::io::Error>
|
||||
where
|
||||
F: FnMut(&mut BufReader<File>, BufWriter<File>) -> Result<u64, std::io::Error>,
|
||||
F: FnMut(&mut BufReader<File>, BufWriter<File>) -> 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 `<archive>.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<u32> = 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,
|
||||
|
||||
+143
-26
@@ -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<FileInfo> {
|
||||
/// 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<FileInfo> {
|
||||
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),
|
||||
}
|
||||
|
||||
@@ -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(())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 `*.<ext>.tmp` files the scanner never recognizes.
|
||||
fn compressed_tmp_suffix_len(filename: &str) -> Option<usize> {
|
||||
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<Duration>, 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));
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user