gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-19 02:56:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(obs): optimize logging with custom RollingAppender and improved cleanup (#2151)

Browse Source 
Signed-off-by: houseme <housemecn@gmail.com>
Signed-off-by: heihutu <30542132+heihutu@users.noreply.github.com>
Co-authored-by: Copilot <198982749+Copilot@users.noreply.github.com>
Co-authored-by: houseme <4829346+houseme@users.noreply.github.com>
Co-authored-by: heihutu <30542132+heihutu@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
houseme
2026-03-13 13:20:27 +08:00
committed by GitHub
parent f83bf95b04
commit 593a58c161
15 changed files with 1227 additions and 610 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/obs/Cargo.toml
+1 -1
View File
@@ -38,6 +38,7 @@ rustfs-config = { workspace = true, features = ["constants", "observability"] }
rustfs-utils = { workspace = true, features = ["ip", "path"] }
flate2 = { workspace = true }
glob = { workspace = true }
jiff = { workspace = true }
metrics = { workspace = true }
nvml-wrapper = { workspace = true, optional = true }
opentelemetry = { workspace = true }
@@ -56,7 +57,6 @@ tracing-subscriber = { workspace = true, features = ["registry", "std", "fmt", "
tokio = { workspace = true, features = ["sync", "fs", "rt-multi-thread", "rt", "time", "macros"] }
sysinfo = { workspace = true }
thiserror = { workspace = true }
walkdir = { workspace = true }
[target.'cfg(unix)'.dependencies]
pyroscope = { workspace = true, features = ["backend-pprof-rs"] }
crates/obs/README.md
+22 -17
View File
@@ -1,7 +1,7 @@
# rustfs-obs
Observability library for [RustFS](https://github.com/rustfs/rustfs) providing structured JSON
logging, distributed tracing, and metrics via OpenTelemetry.
logging, distributed tracing, metrics via OpenTelemetry, and continuous profiling via Pyroscope.
---
@@ -10,9 +10,10 @@ logging, distributed tracing, and metrics via OpenTelemetry.
| Feature | Description |
|---------|-------------|
| **Structured logging** | JSON-formatted logs via `tracing-subscriber` |
| **Rolling-file logging** | Daily / hourly rotation with automatic cleanup |
| **Rolling-file logging** | Daily / hourly rotation with automatic cleanup and high-precision timestamps |
| **Distributed tracing** | OTLP/HTTP export to Jaeger, Tempo, or any OTel collector |
| **Metrics** | OTLP/HTTP export, bridged from the `metrics` crate facade |
| **Continuous Profiling** | CPU/Memory profiling export to Pyroscope |
| **Log cleanup** | Background task: size limits, gzip compression, retention policies |
| **GPU metrics** *(optional)* | Enable with the `gpu` feature flag |
@@ -79,7 +80,7 @@ The library selects a backend automatically based on configuration:
```
1. Any OTLP endpoint set?
└─ YES → Full OTLP/HTTP pipeline (traces + metrics + logs)
└─ YES → Full OTLP/HTTP pipeline (traces + metrics + logs + profiling)
2. RUSTFS_OBS_LOG_DIRECTORY set to a non-empty path?
└─ YES → Rolling-file JSON logging
@@ -110,9 +111,11 @@ All configuration is read from environment variables at startup.
| `RUSTFS_OBS_TRACE_ENDPOINT` | _(empty)_ | Dedicated trace endpoint (overrides root + `/v1/traces`) |
| `RUSTFS_OBS_METRIC_ENDPOINT` | _(empty)_ | Dedicated metrics endpoint |
| `RUSTFS_OBS_LOG_ENDPOINT` | _(empty)_ | Dedicated log endpoint |
| `RUSTFS_OBS_PROFILING_ENDPOINT` | _(empty)_ | Dedicated profiling endpoint (e.g. Pyroscope) |
| `RUSTFS_OBS_TRACES_EXPORT_ENABLED` | `true` | Toggle trace export |
| `RUSTFS_OBS_METRICS_EXPORT_ENABLED` | `true` | Toggle metrics export |
| `RUSTFS_OBS_LOGS_EXPORT_ENABLED` | `true` | Toggle OTLP log export |
| `RUSTFS_OBS_PROFILING_EXPORT_ENABLED` | `true` | Toggle profiling export |
| `RUSTFS_OBS_USE_STDOUT` | `false` | Mirror all signals to stdout alongside OTLP |
| `RUSTFS_OBS_SAMPLE_RATIO` | `0.1` | Trace sampling ratio `0.0`–`1.0` |
| `RUSTFS_OBS_METER_INTERVAL` | `15` | Metrics export interval (seconds) |
@@ -132,7 +135,7 @@ All configuration is read from environment variables at startup.
| `RUSTFS_OBS_LOGGER_LEVEL` | `info` | Log level; `RUST_LOG` syntax supported |
| `RUSTFS_OBS_LOG_STDOUT_ENABLED` | `false` | When file logging is active, also mirror to stdout |
| `RUSTFS_OBS_LOG_DIRECTORY` | _(empty)_ | **Directory for rolling log files. When empty, logs go to stdout only** |
| `RUSTFS_OBS_LOG_FILENAME` | `rustfs.log` | Base filename for rolling logs (date suffix added automatically) |
| `RUSTFS_OBS_LOG_FILENAME` | `rustfs.log` | Base filename for rolling logs. Rotated archives include a high-precision timestamp and counter. With the default `RUSTFS_OBS_LOG_MATCH_MODE=suffix`, names look like `<timestamp>-<counter>.rustfs.log` (e.g., `20231027103001.123456-0.rustfs.log`); with `prefix`, they look like `rustfs.log.<timestamp>-<counter>` (e.g., `rustfs.log.20231027103001.123456-0`). |
| `RUSTFS_OBS_LOG_ROTATION_TIME` | `hourly` | Rotation granularity: `minutely`, `hourly`, or `daily` |
| `RUSTFS_OBS_LOG_KEEP_FILES` | `30` | Number of rolling files to keep (also used by cleaner) |
| `RUSTFS_OBS_LOG_MATCH_MODE` | `suffix` | File matching mode: `prefix` or `suffix` |
@@ -247,21 +250,23 @@ use std::path::PathBuf;
use rustfs_obs::LogCleaner;
use rustfs_obs::types::FileMatchMode;
let cleaner = LogCleaner::new(
let cleaner = LogCleaner::builder(
PathBuf::from("/var/log/rustfs"),
"rustfs.log.".to_string(), // file_pattern
FileMatchMode::Prefix, // match_mode
10, // keep_count
2 * 1024 * 1024 * 1024, // max_total_size_bytes (2 GiB)
0, // max_single_file_size_bytes (unlimited)
true, // compress_old_files
6, // gzip_compression_level
30, // compressed_file_retention_days
vec!["current.log".to_string()], // exclude_patterns
true, // delete_empty_files
3600, // min_file_age_seconds (1 hour)
false, // dry_run
);
"rustfs.log".to_string(), // active_filename
)
.match_mode(FileMatchMode::Prefix)
.keep_files(10)
.max_total_size_bytes(2 * 1024 * 1024 * 1024) // 2 GiB
.max_single_file_size_bytes(0) // unlimited
.compress_old_files(true)
.gzip_compression_level(6)
.compressed_file_retention_days(30)
.exclude_patterns(vec!["current.log".to_string()])
.delete_empty_files(true)
.min_file_age_seconds(3600) // 1 hour
.dry_run(false)
.build();
let (deleted, freed_bytes) = cleaner.cleanup().expect("cleanup failed");
println!("Deleted {deleted} files, freed {freed_bytes} bytes");
crates/obs/src/cleaner/README.md
+71
View File
@@ -0,0 +1,71 @@
# Log Cleaner Subsystem
The `cleaner` module provides a robust, background log-file lifecycle manager for RustFS. It is designed to run periodically to enforce retention policies, compress old logs, and prevent disk exhaustion.
## Architecture
The cleaner operates as a pipeline:
1. **Discovery (`scanner.rs`)**: Scans the configured log directory for eligible files.
* **Non-recursive**: Only scans the top-level directory for safety.
* **Filtering**: Ignores the currently active log file, files matching exclude patterns, and files that do not match the configured prefix/suffix pattern.
* **Performance**: Uses `std::fs::read_dir` directly to minimize overhead and syscalls.
2. **Selection (`core.rs`)**: Applies retention policies to select files for deletion.
* **Keep Count**: Ensures at least `N` recent files are kept.
* **Total Size**: Deletes oldest files if the total size exceeds the limit.
* **Single File Size**: Deletes individual files that exceed a size limit (e.g., runaway logs).
3. **Action (`core.rs` / `compress.rs`)**:
* **Compression**: Optionally compresses selected files using Gzip (level 1-9) before deletion.
* **Deletion**: Removes the original file (and eventually the compressed archive based on retention days).
## Configuration
The cleaner is configured via `LogCleanerBuilder`. When initialized via `rustfs-obs::init_obs`, it reads from environment variables.
| Parameter | Env Var | Description |
|-----------|---------|-------------|
| `log_dir` | `RUSTFS_OBS_LOG_DIRECTORY` | The directory to scan. |
| `file_pattern` | `RUSTFS_OBS_LOG_FILENAME` | The base filename pattern (e.g., `rustfs.log`). |
| `active_filename` | (Derived) | The exact name of the currently active log file, excluded from cleanup. |
| `match_mode` | `RUSTFS_OBS_LOG_MATCH_MODE` | `prefix` or `suffix`. Determines how `file_pattern` is matched against filenames. |
| `keep_files` | `RUSTFS_OBS_LOG_KEEP_FILES` | Minimum number of rolling log files to keep. |
| `max_total_size_bytes` | `RUSTFS_OBS_LOG_MAX_TOTAL_SIZE_BYTES` | Maximum aggregate size of all log files. Oldest files are deleted to satisfy this. |
| `compress_old_files` | `RUSTFS_OBS_LOG_COMPRESS_OLD_FILES` | If `true`, files selected for removal are first gzipped. |
| `compressed_file_retention_days` | `RUSTFS_OBS_LOG_COMPRESSED_FILE_RETENTION_DAYS` | Age in days after which `.gz` files are deleted. |
## Timestamp Format & Rotation
The cleaner works in tandem with the `RollingAppender` in `telemetry/rolling.rs`.
* **Rotation**: Logs are rotated based on time (Daily/Hourly/Minutely) or Size.
* **Naming**: Archived logs use a high-precision timestamp format: `YYYYMMDDHHMMSS.uuuuuu` (microseconds), plus a unique counter to prevent collisions.
* **Suffix Mode**: `<timestamp>-<counter>.<filename>` (e.g., `20231027103001.123456-0.rustfs.log`)
* **Prefix Mode**: `<filename>.<timestamp>-<counter>` (e.g., `rustfs.log.20231027103001.123456-0`)
This high-precision naming ensures that files sort chronologically by name, and collisions are virtually impossible even under high load.
## Usage Example
```rust
use rustfs_obs::LogCleaner;
use rustfs_obs::types::FileMatchMode;
use std::path::PathBuf;
let cleaner = LogCleaner::builder(
PathBuf::from("/var/log/rustfs"),
"rustfs.log.".to_string(),
"rustfs.log".to_string(),
)
.match_mode(FileMatchMode::Prefix)
.keep_files(10)
.max_total_size_bytes(1024 * 1024 * 100) // 100 MB
.compress_old_files(true)
.build();
// Run cleanup (blocking operation, spawn in a background task)
if let Ok((deleted, freed)) = cleaner.cleanup() {
println!("Cleaned up {} files, freed {} bytes", deleted, freed);
}
```
crates/obs/src/cleaner/core.rs
+243 -198
View File
@@ -15,7 +15,7 @@
//! Core log-file cleanup orchestration.
//!
//! [`LogCleaner`] is the public entry point for the cleanup subsystem.
//! Construct it with [`LogCleaner::new`] and call [`LogCleaner::cleanup`]
//! Construct it with [`LogCleaner::builder`] and call [`LogCleaner::cleanup`]
//! periodically (e.g. from a `tokio::spawn`-ed loop).
//!
//! Internally the cleaner delegates to:
@@ -24,9 +24,11 @@
//! - [`LogCleaner::select_files_to_delete`] — to apply count / size limits.
use super::compress::compress_file;
use super::scanner::{collect_expired_compressed_files, collect_log_files};
use super::scanner::{LogScanResult, scan_log_directory};
use super::types::{FileInfo, FileMatchMode};
use rustfs_config::DEFAULT_LOG_KEEP_FILES;
use std::path::PathBuf;
use std::time::SystemTime;
use tracing::{debug, error, info};
/// Log-file lifecycle manager.
@@ -43,6 +45,8 @@ pub struct LogCleaner {
pub(super) log_dir: PathBuf,
/// Pattern string to match files (used as prefix or suffix).
pub(super) file_pattern: String,
/// Exact name of the active log file (to exclude from cleanup).
pub(super) active_filename: String,
/// Whether to match by prefix or suffix.
pub(super) match_mode: FileMatchMode,
/// The cleaner will never delete files if doing so would leave fewer than
@@ -70,54 +74,19 @@ pub struct LogCleaner {
}
impl LogCleaner {
/// Build a new [`LogCleaner`] with the supplied policy parameters.
///
/// `exclude_patterns` is a list of glob strings (e.g. `"*.lock"`). Invalid
/// glob patterns are silently ignored.
///
/// `gzip_compression_level` is clamped to the range `[1, 9]`.
#[allow(clippy::too_many_arguments)]
pub fn new(
log_dir: PathBuf,
file_pattern: String,
match_mode: FileMatchMode,
keep_files: usize,
max_total_size_bytes: u64,
max_single_file_size_bytes: u64,
compress_old_files: bool,
gzip_compression_level: u32,
compressed_file_retention_days: u64,
exclude_patterns: Vec<String>,
delete_empty_files: bool,
min_file_age_seconds: u64,
dry_run: bool,
) -> Self {
let patterns = exclude_patterns
.into_iter()
.filter_map(|p| glob::Pattern::new(&p).ok())
.collect();
Self {
log_dir,
file_pattern,
match_mode,
keep_files,
max_total_size_bytes,
max_single_file_size_bytes,
compress_old_files,
gzip_compression_level: gzip_compression_level.clamp(1, 9),
compressed_file_retention_days,
exclude_patterns: patterns,
delete_empty_files,
min_file_age_seconds,
dry_run,
}
/// Create a builder to construct a `LogCleaner`.
pub fn builder(
log_dir: impl Into<PathBuf>,
file_pattern: impl Into<String>,
active_filename: impl Into<String>,
) -> LogCleanerBuilder {
LogCleanerBuilder::new(log_dir, file_pattern, active_filename)
}
/// Perform one full cleanup pass.
///
/// Steps:
/// 1. Scan the log directory for managed files.
/// 1. Scan the log directory for managed files (excluding the active file).
/// 2. Apply count/size policies to select files for deletion.
/// 3. Optionally compress selected files, then delete them.
/// 4. Collect and delete expired compressed archives.
@@ -137,10 +106,15 @@ impl LogCleaner {
let mut total_deleted = 0usize;
let mut total_freed = 0u64;
// ── 1. Discover active log files ──────────────────────────────────────
let mut files = collect_log_files(
// ── 1. Discover active log files (Archives only) ──────────────────────
// We explicitly pass `active_filename` to exclude it from the list.
let LogScanResult {
mut logs,
mut compressed_archives,
} = scan_log_directory(
&self.log_dir,
&self.file_pattern,
Some(&self.active_filename),
self.match_mode,
&self.exclude_patterns,
self.min_file_age_seconds,
@@ -148,27 +122,19 @@ impl LogCleaner {
self.dry_run,
)?;
if files.is_empty() {
debug!("No log files found in directory: {:?}", self.log_dir);
} else {
files.sort_by_key(|f| f.modified);
let total_size: u64 = files.iter().map(|f| f.size).sum();
// ── 2. Select + compress + delete (Regular Logs) ──────────────────────
if !logs.is_empty() {
logs.sort_by_key(|f| f.modified);
let total_size: u64 = logs.iter().map(|f| f.size).sum();
info!(
"Found {} log files, total size: {} bytes ({:.2} MB)",
files.len(),
"Found {} regular log files, total size: {} bytes ({:.2} MB)",
logs.len(),
total_size,
total_size as f64 / 1024.0 / 1024.0
);
// ── 2. Select + compress + delete ─────────────────────────────────
let (to_delete, to_rotate) = self.select_files_to_process(&files, total_size);
// Handle rotation for active file if needed
if let Some(active_file) = to_rotate
&& let Err(e) = self.rotate_active_file(&active_file)
{
error!("Failed to rotate active file {:?}: {}", active_file.path, e);
}
let to_delete = self.select_files_to_process(&logs, total_size);
if !to_delete.is_empty() {
let (d, f) = self.compress_and_delete(&to_delete)?;
@@ -178,16 +144,13 @@ impl LogCleaner {
}
// ── 3. Remove expired compressed archives ─────────────────────────────
let expired_gz = collect_expired_compressed_files(
&self.log_dir,
&self.file_pattern,
self.match_mode,
self.compressed_file_retention_days,
)?;
if !expired_gz.is_empty() {
let (d, f) = self.delete_files(&expired_gz)?;
total_deleted += d;
total_freed += f;
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)?;
total_deleted += d;
total_freed += f;
}
}
if total_deleted > 0 || total_freed > 0 {
@@ -204,28 +167,19 @@ impl LogCleaner {
// ─── Selection ────────────────────────────────────────────────────────────
/// Choose which files from `files` (sorted oldest-first) should be deleted or rotated.
/// Choose which files from `files` (sorted oldest-first) should be deleted.
///
/// The algorithm respects three constraints in order:
/// 1. Always keep at least `keep_files` files.
/// 1. Always keep at least `keep_files` files (archives).
/// 2. Delete old files while the total size exceeds `max_total_size_bytes`.
/// 3. Delete any file whose individual size exceeds `max_single_file_size_bytes`.
///
/// **Note**: The most recent file (assumed to be the active log) is exempt
/// from size-based deletion. If it exceeds the size limit, it is returned
/// as `to_rotate`.
pub(super) fn select_files_to_process(&self, files: &[FileInfo], total_size: u64) -> (Vec<FileInfo>, Option<FileInfo>) {
pub(super) fn select_files_to_process(&self, files: &[FileInfo], total_size: u64) -> Vec<FileInfo> {
let mut to_delete = Vec::new();
let mut to_rotate = None;
if files.is_empty() {
return (to_delete, to_rotate);
return to_delete;
}
// Identify the index of the most recent file (last in the sorted list).
// We will protect this file from size-based deletion.
let active_file_idx = files.len() - 1;
// Calculate how many files we *must* delete to satisfy keep_files.
let must_delete_count = files.len().saturating_sub(self.keep_files);
@@ -244,142 +198,111 @@ impl LogCleaner {
let over_total = self.max_total_size_bytes > 0 && current_size > self.max_total_size_bytes;
// Condition 3: Enforce max_single_file_size_bytes.
// Note: Since active file is excluded, if an archive is > max_single, it means it
// was rotated out being too large (likely) or we lowered the limit. It should be deleted.
let over_single = self.max_single_file_size_bytes > 0 && file.size > self.max_single_file_size_bytes;
if over_total {
// If we are over total size, we delete unless it's the active file.
if idx == active_file_idx {
debug!(
"Active log file contributes to total size limit overflow, but skipping deletion to preserve current logs."
);
} else {
current_size = current_size.saturating_sub(file.size);
to_delete.push(file.clone());
}
current_size = current_size.saturating_sub(file.size);
to_delete.push(file.clone());
} else if over_single {
// For single file limits, we MUST NOT delete the active file.
if idx == active_file_idx {
// Mark active file for rotation instead of deletion
to_rotate = Some(file.clone());
} else {
debug!(
"File exceeds single-file size limit: {:?} ({} > {} bytes)",
file.path, file.size, self.max_single_file_size_bytes
);
current_size = current_size.saturating_sub(file.size);
to_delete.push(file.clone());
}
debug!(
"Archive exceeds single-file size limit: {:?} ({} > {} bytes). Deleting.",
file.path, file.size, self.max_single_file_size_bytes
);
current_size = current_size.saturating_sub(file.size);
to_delete.push(file.clone());
}
}
(to_delete, to_rotate)
to_delete
}
// ─── Rotation ─────────────────────────────────────────────────────────────
/// Select compressed files that have exceeded the retention period.
fn select_expired_compressed(&self, files: &mut [FileInfo]) -> Vec<FileInfo> {
let retention = std::time::Duration::from_secs(self.compressed_file_retention_days * 24 * 3600);
let now = SystemTime::now();
let mut expired = Vec::new();
/// Rotate the active file by renaming it with a timestamp suffix.
/// The original filename will be recreated by the logging appender on next write.
fn rotate_active_file(&self, file: &FileInfo) -> Result<(), std::io::Error> {
if self.dry_run {
info!("[DRY RUN] Would rotate active file: {:?} ({} bytes)", file.path, file.size);
return Ok(());
}
// Generate timestamp: unix timestamp in seconds
let timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_err(std::io::Error::other)?
.as_secs();
let file_name = file
.path
.file_name()
.and_then(|n| n.to_str())
.ok_or_else(|| std::io::Error::new(std::io::ErrorKind::InvalidData, "Invalid filename"))?;
// Construct the rotated filename.
// We must ensure the new filename still matches the file_pattern so it can be discovered
// by the scanner in future runs (and eventually deleted).
//
// Suffix mode: Insert timestamp BEFORE the suffix.
// Example: "2026-03-01.rustfs.log" (pattern="rustfs.log")
// -> "2026-03-01.1740810000.rustfs.log"
//
// Prefix mode: Append timestamp at the end.
// Example: "app.log" (pattern="app")
// -> "app.log.1740810000"
let rotated_name = match self.match_mode {
FileMatchMode::Suffix => {
if let Some(base) = file_name.strip_suffix(&self.file_pattern) {
let mut new_name = String::with_capacity(file_name.len() + 20);
new_name.push_str(base);
// Ensure separator between base and timestamp
if !base.is_empty() && !base.ends_with('.') {
new_name.push('.');
}
new_name.push_str(&timestamp.to_string());
// Ensure separator between timestamp and suffix
if !self.file_pattern.starts_with('.') {
new_name.push('.');
}
new_name.push_str(&self.file_pattern);
new_name
} else {
// Should not happen if scanner works correctly, but fallback safely
format!("{}.{}", file_name, timestamp)
}
for file in files {
if let Ok(age) = now.duration_since(file.modified)
&& age > retention
{
expired.push(file.clone());
}
FileMatchMode::Prefix => {
// For prefix matching, appending to the end preserves the prefix.
format!("{}.{}", file_name, timestamp)
}
};
let rotated_path = file.path.with_file_name(&rotated_name);
// Check if target already exists to avoid overwriting (unlikely with timestamp but possible)
if rotated_path.exists() {
return Err(std::io::Error::new(
std::io::ErrorKind::AlreadyExists,
format!("Rotated file already exists: {:?}", rotated_path),
));
}
info!("Rotating active log file: {:?} -> {:?}", file.path, rotated_path);
// Rename the current active file to the rotated name.
// The logging appender (tracing-appender) will automatically create a new file
// with the original name when it next attempts to write.
// Note: On Linux/Unix, this rename is atomic and safe even if the file is open.
if let Err(e) = std::fs::rename(&file.path, &rotated_path) {
// Add context to the error
return Err(std::io::Error::new(
e.kind(),
format!("Failed to rename {:?} to {:?}: {}", file.path, rotated_path, e),
));
}
Ok(())
expired
}
// ─── Compression + deletion ───────────────────────────────────────────────
/// Securely delete a file, preventing symlink attacks (TOCTOU).
///
/// This function verifies that the path is not a symlink before attempting deletion.
/// While strictly speaking a race condition is still theoretically possible between
/// `symlink_metadata` and `remove_file`, this check covers the vast majority of
/// privilege escalation vectors where a user replaces a log file with a symlink
/// to a system file.
fn secure_delete(&self, path: &PathBuf) -> std::io::Result<()> {
// 1. Lstat (symlink_metadata) - do not follow links
let meta = std::fs::symlink_metadata(path)?;
// 2. Symlink Check
// If it's a symlink, we NEVER delete it. It might point to /etc/passwd.
// In a log directory, symlinks are unexpected and dangerous.
if meta.file_type().is_symlink() {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!("Security: refusing to delete symlink: {:?}", path),
));
}
// 3. Perform Deletion
std::fs::remove_file(path)
}
/// Optionally compress and then delete the given files.
///
/// This function is synchronous and blocking. It should be called within a
/// `spawn_blocking` task if running in an async context.
fn compress_and_delete(&self, files: &[FileInfo]) -> Result<(usize, u64), std::io::Error> {
if self.compress_old_files {
for f in files {
if let Err(e) = compress_file(&f.path, self.gzip_compression_level, self.dry_run) {
tracing::warn!("Failed to compress {:?}: {}", f.path, e);
let mut total_deleted = 0;
let mut total_freed = 0;
for f in files {
let mut deleted_size = 0;
if self.compress_old_files {
match compress_file(&f.path, self.gzip_compression_level, self.dry_run) {
Ok(_) => {}
Err(e) => {
tracing::warn!("Failed to compress {:?}: {}", f.path, e);
}
}
}
// Now delete
if self.dry_run {
info!("[DRY RUN] Would delete: {:?} ({} bytes)", f.path, f.size);
deleted_size = f.size;
} else {
match self.secure_delete(&f.path) {
Ok(()) => {
debug!("Deleted: {:?}", f.path);
deleted_size = f.size;
}
Err(e) => {
error!("Failed to delete {:?}: {}", f.path, e);
}
}
}
if deleted_size > 0 {
total_deleted += 1;
total_freed += deleted_size;
}
}
self.delete_files(files)
Ok((total_deleted, total_freed))
}
/// Delete all files in `files`, logging each operation.
@@ -398,7 +321,7 @@ impl LogCleaner {
deleted += 1;
freed += f.size;
} else {
match std::fs::remove_file(&f.path) {
match self.secure_delete(&f.path) {
Ok(()) => {
debug!("Deleted: {:?}", f.path);
deleted += 1;
@@ -414,3 +337,125 @@ impl LogCleaner {
Ok((deleted, freed))
}
}
/// Builder for [`LogCleaner`].
pub struct LogCleanerBuilder {
log_dir: PathBuf,
file_pattern: String,
active_filename: String,
match_mode: FileMatchMode,
keep_files: usize,
max_total_size_bytes: u64,
max_single_file_size_bytes: u64,
compress_old_files: bool,
gzip_compression_level: u32,
compressed_file_retention_days: u64,
exclude_patterns: Vec<String>,
delete_empty_files: bool,
min_file_age_seconds: u64,
dry_run: bool,
}
impl LogCleanerBuilder {
pub fn new(log_dir: impl Into<PathBuf>, file_pattern: impl Into<String>, active_filename: impl Into<String>) -> Self {
Self {
log_dir: log_dir.into(),
file_pattern: file_pattern.into(),
active_filename: active_filename.into(),
match_mode: FileMatchMode::Prefix,
// Default to a safe non-zero value so that a builder created
// without an explicit `keep_files()` call does not immediately
// delete all matching log files.
keep_files: DEFAULT_LOG_KEEP_FILES,
max_total_size_bytes: 0,
max_single_file_size_bytes: 0,
compress_old_files: false,
gzip_compression_level: 6,
compressed_file_retention_days: 0,
exclude_patterns: Vec::new(),
delete_empty_files: false,
min_file_age_seconds: 0,
dry_run: false,
}
}
pub fn match_mode(mut self, match_mode: FileMatchMode) -> Self {
self.match_mode = match_mode;
self
}
pub fn keep_files(mut self, keep_files: usize) -> Self {
self.keep_files = keep_files;
self
}
pub fn max_total_size_bytes(mut self, max_total_size_bytes: u64) -> Self {
self.max_total_size_bytes = max_total_size_bytes;
self
}
pub fn max_single_file_size_bytes(mut self, max_single_file_size_bytes: u64) -> Self {
self.max_single_file_size_bytes = max_single_file_size_bytes;
self
}
pub fn compress_old_files(mut self, compress_old_files: bool) -> Self {
self.compress_old_files = compress_old_files;
self
}
pub fn gzip_compression_level(mut self, gzip_compression_level: u32) -> Self {
self.gzip_compression_level = gzip_compression_level;
self
}
pub fn compressed_file_retention_days(mut self, days: u64) -> Self {
self.compressed_file_retention_days = days;
self
}
pub fn exclude_patterns(mut self, patterns: Vec<String>) -> Self {
self.exclude_patterns = patterns;
self
}
pub fn delete_empty_files(mut self, delete_empty_files: bool) -> Self {
self.delete_empty_files = delete_empty_files;
self
}
pub fn min_file_age_seconds(mut self, seconds: u64) -> Self {
self.min_file_age_seconds = seconds;
self
}
pub fn dry_run(mut self, dry_run: bool) -> Self {
self.dry_run = dry_run;
self
}
pub fn build(self) -> LogCleaner {
let patterns = self
.exclude_patterns
.into_iter()
.filter_map(|p| glob::Pattern::new(&p).ok())
.collect();
LogCleaner {
log_dir: self.log_dir,
file_pattern: self.file_pattern,
active_filename: self.active_filename,
match_mode: self.match_mode,
keep_files: self.keep_files,
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),
compressed_file_retention_days: self.compressed_file_retention_days,
exclude_patterns: patterns,
delete_empty_files: self.delete_empty_files,
min_file_age_seconds: self.min_file_age_seconds,
dry_run: self.dry_run,
}
}
}
crates/obs/src/cleaner/mod.rs
+41 -62
View File
@@ -33,21 +33,23 @@
//! use rustfs_obs::LogCleaner;
//! use rustfs_obs::types::FileMatchMode;
//!
//! let cleaner = LogCleaner::new(
//! let cleaner = LogCleaner::builder(
//! PathBuf::from("/var/log/rustfs"),
//! "rustfs.log.".to_string(),
//! FileMatchMode::Prefix,
//! 10, // keep_files
//! 2 * 1024 * 1024 * 1024, // max_total_size_bytes (2 GiB)
//! 0, // max_single_file_size_bytes (unlimited)
//! true, // compress_old_files
//! 6, // gzip_compression_level
//! 30, // compressed_file_retention_days
//! vec![], // exclude_patterns
//! true, // delete_empty_files
//! 3600, // min_file_age_seconds (1 hour)
//! false, // dry_run
//! );
//! "rustfs.log".to_string(),
//! )
//! .match_mode(FileMatchMode::Prefix)
//! .keep_files(10)
//! .max_total_size_bytes(2 * 1024 * 1024 * 1024) // 2 GiB
//! .max_single_file_size_bytes(0) // unlimited
//! .compress_old_files(true)
//! .gzip_compression_level(6)
//! .compressed_file_retention_days(30)
//! .exclude_patterns(vec![])
//! .delete_empty_files(true)
//! .min_file_age_seconds(3600) // 1 hour
//! .dry_run(false)
//! .build();
//!
//! let (deleted, freed_bytes) = cleaner.cleanup().expect("cleanup failed");
//! println!("Deleted {deleted} files, freed {freed_bytes} bytes");
@@ -79,21 +81,13 @@ mod tests {
/// Build a cleaner with sensible test defaults (no compression, no age gate).
fn make_cleaner(dir: std::path::PathBuf, keep: usize, max_bytes: u64) -> LogCleaner {
LogCleaner::new(
dir,
"app.log.".to_string(),
FileMatchMode::Prefix,
keep,
max_bytes,
0, // max_single_file_size_bytes
false, // compress_old_files
6, // gzip_compression_level
30, // compressed_file_retention_days
Vec::new(), // exclude_patterns
true, // delete_empty_files
0, // min_file_age_seconds (0 = no age gate in tests)
false, // dry_run
)
LogCleaner::builder(dir, "app.log.".to_string(), "app.log".to_string())
.match_mode(FileMatchMode::Prefix)
.keep_files(keep)
.max_total_size_bytes(max_bytes)
.min_file_age_seconds(0) // 0 = no age gate in tests
.delete_empty_files(true)
.build()
}
#[test]
@@ -141,11 +135,15 @@ mod tests {
create_log_file(&dir, "app.log.2024-01-02", 1024)?;
create_log_file(&dir, "other.log", 512)?; // different prefix
let cleaner = make_cleaner(dir.clone(), 1, 512);
// keep_files=1 and max_bytes=1500: deleting one managed file (1024 bytes) leaves
// a single managed file of 1024 bytes, which satisfies both the file-count and
// size limits. "other.log" (different prefix) must never be touched.
let cleaner = make_cleaner(dir.clone(), 1, 1500);
let (deleted, _) = cleaner.cleanup()?;
// "other.log" must not be counted or deleted.
// "other.log" must not be counted or deleted; only 1 managed file removed.
assert_eq!(deleted, 1, "only managed files should be deleted");
assert!(dir.join("other.log").exists(), "unrelated file must not be deleted");
Ok(())
}
@@ -158,8 +156,8 @@ mod tests {
create_log_file(&dir, "app.log.2024-01-02", 2048)?;
create_log_file(&dir, "other.log", 512)?;
let files = scanner::collect_log_files(&dir, "app.log.", FileMatchMode::Prefix, &[], 0, true, false)?;
assert_eq!(files.len(), 2, "scanner should find exactly 2 managed files");
let result = scanner::scan_log_directory(&dir, "app.log.", Some("app.log"), FileMatchMode::Prefix, &[], 0, true, false)?;
assert_eq!(result.logs.len(), 2, "scanner should find exactly 2 managed files");
Ok(())
}
@@ -172,21 +170,12 @@ mod tests {
create_log_file(&dir, "app.log.2024-01-02", 1024)?;
create_log_file(&dir, "app.log.2024-01-03", 1024)?;
let cleaner = LogCleaner::new(
dir.clone(),
"app.log.".to_string(),
FileMatchMode::Prefix,
1,
1024,
0,
false,
6,
30,
vec![],
true,
0,
true,
);
let cleaner = LogCleaner::builder(dir.clone(), "app.log.".to_string(), "app.log".to_string())
.match_mode(FileMatchMode::Prefix)
.keep_files(1)
.max_total_size_bytes(1024)
.dry_run(true)
.build();
let (deleted, _freed) = cleaner.cleanup()?;
// dry_run=true reports deletions but doesn't actually remove files.
@@ -204,21 +193,11 @@ mod tests {
create_log_file(&dir, "2026-03-01-06-22.rustfs.log", 1024)?;
create_log_file(&dir, "other.log", 1024)?; // not managed
let cleaner = LogCleaner::new(
dir.clone(),
"rustfs.log".to_string(),
FileMatchMode::Suffix,
1,
1024,
0,
false,
6,
30,
vec![],
true,
0,
false,
);
let cleaner = LogCleaner::builder(dir.clone(), ".rustfs.log".to_string(), "current.log".to_string())
.match_mode(FileMatchMode::Suffix)
.keep_files(1)
.max_total_size_bytes(1024)
.build();
let (deleted, freed) = cleaner.cleanup()?;
assert_eq!(deleted, 1, "should delete exactly one file");
crates/obs/src/cleaner/scanner.rs
+112 -131
View File
@@ -14,55 +14,91 @@
//! Filesystem scanner for discovering log files eligible for cleanup.
//!
//! This module is intentionally kept read-only: it does **not** delete or
//! compress any files — it only reports what it found.
//! This module is primarily read-only: it reports what files it found.
//! The one exception is zero-byte file removal — when `delete_empty_files`
//! is enabled, `scan_log_directory` removes empty regular files as part of
//! the scan so that they are not counted in retention calculations.
use super::types::{FileInfo, FileMatchMode};
use rustfs_config::observability::DEFAULT_OBS_LOG_GZIP_COMPRESSION_ALL_EXTENSION;
use std::fs;
use std::path::Path;
use std::time::{Duration, SystemTime};
use std::time::SystemTime;
use tracing::debug;
use walkdir::WalkDir;
/// Collect all log files in `log_dir` whose name matches `file_pattern` based on `match_mode`.
/// Result of a single pass directory scan.
pub(super) struct LogScanResult {
/// Regular log files eligible for deletion/compression.
pub logs: Vec<FileInfo>,
/// Already compressed files eligible for expiry deletion.
pub compressed_archives: Vec<FileInfo>,
}
/// Perform a single-pass scan of the log directory.
///
/// Files that:
/// - are already compressed (`.gz` extension),
/// - are zero-byte and `delete_empty_files` is `true` (these are handled
/// immediately by the caller), or
/// - match one of the `exclude_patterns`,
/// - were modified more recently than `min_file_age_seconds` seconds ago,
///
/// are skipped and not returned in the result list.
/// This function iterates over the directory entries once and categorizes them
/// into regular logs or compressed archives based on extensions and patterns.
///
/// # Arguments
/// * `log_dir` - Root directory to scan (depth 1 only, no recursion).
/// * `file_pattern` - Pattern string to match filenames.
/// * `active_filename` - The name of the currently active log file (to be excluded).
/// * `match_mode` - Whether to match by prefix or suffix.
/// * `exclude_patterns` - Compiled glob patterns; matching files are skipped.
/// * `min_file_age_seconds` - Files younger than this threshold are skipped.
/// * `delete_empty_files` - When `true`, zero-byte files trigger an immediate
/// delete by the caller before the rest of cleanup runs.
pub(super) fn collect_log_files(
/// * `min_file_age_seconds` - Files younger than this threshold are skipped (for regular logs).
/// * `delete_empty_files` - When `true`, zero-byte regular files that match
/// the pattern are deleted immediately inside this function and excluded
/// from the returned [`LogScanResult`].
#[allow(clippy::too_many_arguments)]
pub(super) fn scan_log_directory(
log_dir: &Path,
file_pattern: &str,
active_filename: Option<&str>,
match_mode: FileMatchMode,
exclude_patterns: &[glob::Pattern],
min_file_age_seconds: u64,
delete_empty_files: bool,
dry_run: bool,
) -> Result<Vec<FileInfo>, std::io::Error> {
let mut files = Vec::new();
) -> Result<LogScanResult, std::io::Error> {
let mut logs = Vec::new();
let mut compressed_archives = Vec::new();
let now = SystemTime::now();
for entry in WalkDir::new(log_dir)
.max_depth(1)
.follow_links(false)
.into_iter()
.filter_map(|e| e.ok())
{
// Use read_dir for a lightweight, non-recursive scan.
let entries = match fs::read_dir(log_dir) {
Ok(entries) => entries,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
// If the log directory does not exist (or was removed), treat this
// as "no files found" instead of failing the whole cleanup pass.
return Ok(LogScanResult {
logs,
compressed_archives,
});
}
Err(e) => return Err(e),
};
for entry in entries {
let entry = match entry {
Ok(e) => e,
Err(_) => continue, // Skip unreadable entries
};
let path = entry.path();
if !path.is_file() {
// We only care about regular files inside the log directory.
// Use `fs::symlink_metadata` (which does *not* follow symlinks) for
// both the file-type check *and* size/mtime collection below. Using
// `entry.metadata()` or `Path::is_file()` (both of which follow
// symlinks) would allow a symlink placed in the log directory to reach
// files outside the tree, and would introduce a TOCTOU window between
// the type-check and the metadata read.
let metadata = match fs::symlink_metadata(&path) {
Ok(md) => md,
Err(_) => continue,
};
let file_type = metadata.file_type();
if !file_type.is_file() {
continue;
}
@@ -71,42 +107,53 @@ pub(super) fn collect_log_files(
None => continue,
};
// Match filename based on mode
// 1. Explicitly skip the active log file (if known).
if let Some(active) = active_filename
&& filename == active
{
continue;
}
// 2. Check exclusion patterns early.
if is_excluded(filename, exclude_patterns) {
debug!("Excluding file from cleanup: {:?}", filename);
continue;
}
// 3. Classify file type and check pattern match.
let is_compressed = filename.ends_with(DEFAULT_OBS_LOG_GZIP_COMPRESSION_ALL_EXTENSION);
// For matching, we need the "base" name.
// If compressed: "foo.log.gz" -> check "foo.log"
// If regular: "foo.log" -> check "foo.log"
let name_to_match = if is_compressed {
&filename[..filename.len() - DEFAULT_OBS_LOG_GZIP_COMPRESSION_ALL_EXTENSION.len()]
} else {
filename
};
let matches = match match_mode {
FileMatchMode::Prefix => filename.starts_with(file_pattern),
FileMatchMode::Suffix => filename.ends_with(file_pattern),
FileMatchMode::Prefix => name_to_match.starts_with(file_pattern),
FileMatchMode::Suffix => name_to_match.ends_with(file_pattern),
};
if !matches {
continue;
}
// Compressed files are handled by collect_compressed_files.
if filename.ends_with(DEFAULT_OBS_LOG_GZIP_COMPRESSION_ALL_EXTENSION) {
continue;
}
// Honour exclusion patterns.
if is_excluded(filename, exclude_patterns) {
debug!("Excluding file from cleanup: {:?}", filename);
continue;
}
let metadata = match entry.metadata() {
Ok(m) => m,
Err(_) => continue,
};
// 4. Gather size and mtime from the already-fetched symlink_metadata
// (reuse; no second syscall, no symlink following).
let file_size = metadata.len();
let modified = match metadata.modified() {
Ok(t) => t,
Err(_) => continue,
Err(_) => continue, // Skip files where we can't read modification time
};
let file_size = metadata.len();
// Delete zero-byte files immediately (outside the normal selection
// logic) when the feature is enabled.
if file_size == 0 && delete_empty_files {
// 5. Handle zero-byte files (Regular logs only).
// We generally don't delete empty compressed files implicitly, but let's stick to regular files logic.
if !is_compressed && file_size == 0 && delete_empty_files {
if !dry_run {
if let Err(e) = std::fs::remove_file(path) {
if let Err(e) = std::fs::remove_file(&path) {
tracing::warn!("Failed to delete empty file {:?}: {}", path, e);
} else {
debug!("Deleted empty file: {:?}", path);
@@ -117,99 +164,33 @@ pub(super) fn collect_log_files(
continue;
}
// Skip files that are too young.
if let Ok(age) = now.duration_since(modified)
// 6. Age Check (Regular logs only).
// Compressed files have their own retention check in the caller.
if !is_compressed
&& let Ok(age) = now.duration_since(modified)
&& age.as_secs() < min_file_age_seconds
{
debug!(
"Skipping file (too new): {:?}, age: {}s, min_age: {}s",
filename,
age.as_secs(),
min_file_age_seconds
);
// Too young to be touched.
continue;
}
files.push(FileInfo {
path: path.to_path_buf(),
let info = FileInfo {
path,
size: file_size,
modified,
});
}
Ok(files)
}
/// Collect compressed `.gz` log files whose age exceeds the retention period.
///
/// When `compressed_file_retention_days` is `0` the function returns immediately
/// without collecting anything (files are kept indefinitely).
///
/// # Arguments
/// * `log_dir` - Root directory to scan.
/// * `file_pattern` - Pattern string to match filenames.
/// * `match_mode` - Whether to match by prefix or suffix.
/// * `compressed_file_retention_days` - Files older than this are eligible for
/// deletion; `0` means never delete compressed files.
pub(super) fn collect_expired_compressed_files(
log_dir: &Path,
file_pattern: &str,
match_mode: FileMatchMode,
compressed_file_retention_days: u64,
) -> Result<Vec<FileInfo>, std::io::Error> {
if compressed_file_retention_days == 0 {
return Ok(Vec::new());
}
let retention = Duration::from_secs(compressed_file_retention_days * 24 * 3600);
let now = SystemTime::now();
let mut files = Vec::new();
for entry in WalkDir::new(log_dir)
.max_depth(1)
.follow_links(false)
.into_iter()
.filter_map(|e| e.ok())
{
let path = entry.path();
if !path.is_file() {
continue;
}
let filename = match path.file_name().and_then(|n| n.to_str()) {
Some(f) => f,
None => continue,
};
if !filename.ends_with(DEFAULT_OBS_LOG_GZIP_COMPRESSION_ALL_EXTENSION) {
continue;
}
// Check if the base filename (without .gz) matches the pattern
let base_filename = &filename[..filename.len() - 3];
let matches = match match_mode {
FileMatchMode::Prefix => base_filename.starts_with(file_pattern),
FileMatchMode::Suffix => base_filename.ends_with(file_pattern),
};
if !matches {
continue;
}
let Ok(metadata) = entry.metadata() else { continue };
let Ok(modified) = metadata.modified() else { continue };
let Ok(age) = now.duration_since(modified) else { continue };
if age > retention {
files.push(FileInfo {
path: path.to_path_buf(),
size: metadata.len(),
modified,
});
if is_compressed {
compressed_archives.push(info);
} else {
logs.push(info);
}
}
Ok(files)
Ok(LogScanResult {
logs,
compressed_archives,
})
}
/// Returns `true` if `filename` matches any of the compiled exclusion patterns.
crates/obs/src/config.rs
+5 -1
View File
@@ -210,7 +210,11 @@ impl OtelConfig {
// `log_keep_files` is the single source of truth for file retention count.
// It defaults to `DEFAULT_LOG_KEEP_FILES` (30).
let log_keep_files = Some(get_env_usize(ENV_OBS_LOG_KEEP_FILES, DEFAULT_LOG_KEEP_FILES));
let mut log_keep_files = get_env_usize(ENV_OBS_LOG_KEEP_FILES, DEFAULT_LOG_KEEP_FILES);
if log_keep_files == 0 {
log_keep_files = DEFAULT_LOG_KEEP_FILES;
}
let log_keep_files = Some(log_keep_files);
// `log_rotation_time` drives the rolling-appender rotation period.
let log_rotation_time = Some(get_env_str(ENV_OBS_LOG_ROTATION_TIME, DEFAULT_LOG_ROTATION_TIME));
crates/obs/src/telemetry/filter.rs
+18 -5
View File
@@ -75,12 +75,25 @@ pub(super) fn build_env_filter(logger_level: &str, default_level: Option<&str>)
.map(EnvFilter::new)
.unwrap_or_else(|| EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(level)));
// Suppress chatty infrastructure crates unless the operator explicitly
// requests trace/debug output.
if should_suppress_noisy_crates(logger_level, default_level, rust_log.as_deref()) {
let directives: SmallVec<[&str; 5]> = smallvec::smallvec!["hyper", "tonic", "h2", "reqwest", "tower"];
for directive in directives {
filter = filter.add_directive(format!("{directive}=off").parse().unwrap());
let directives: SmallVec<[(&str, &str); 6]> = smallvec::smallvec![
("hyper", "off"),
("tonic", "off"),
("h2", "off"),
("reqwest", "off"),
("tower", "off"),
// HTTP request logs are demoted to WARN to reduce volume in production.
("rustfs::server::http", "warn"),
];
for (crate_name, level) in directives {
match format!("{crate_name}={level}").parse() {
Ok(directive) => filter = filter.add_directive(directive),
Err(e) => {
// The directive strings are compile-time constants, so this
// branch should never be reached; emit a diagnostic just in case.
eprintln!("obs: invalid log filter directive '{crate_name}={level}': {e}");
}
}
}
}
crates/obs/src/telemetry/local.rs
+39 -40
View File
@@ -32,6 +32,7 @@ use crate::cleaner::types::FileMatchMode;
use crate::config::OtelConfig;
use crate::global::OBSERVABILITY_METRIC_ENABLED;
use crate::telemetry::filter::build_env_filter;
use crate::telemetry::rolling::{RollingAppender, Rotation};
use metrics::counter;
use rustfs_config::observability::{
DEFAULT_OBS_LOG_CLEANUP_INTERVAL_SECONDS, DEFAULT_OBS_LOG_COMPRESS_OLD_FILES, DEFAULT_OBS_LOG_COMPRESSED_FILE_RETENTION_DAYS,
@@ -175,7 +176,7 @@ fn init_file_logging_internal(
// ── 3. Choose rotation strategy ──────────────────────────────────────────
// `log_rotation_time` drives the rolling-appender rotation period.
let rotation = config
let rotation_str = config
.log_rotation_time
.as_deref()
.unwrap_or(DEFAULT_LOG_ROTATION_TIME)
@@ -187,28 +188,20 @@ fn init_file_logging_internal(
_ => FileMatchMode::Suffix,
};
use tracing_appender::rolling::{RollingFileAppender, Rotation};
let file_appender = {
let rotation = match rotation.as_str() {
"minutely" => Rotation::MINUTELY,
"hourly" => Rotation::HOURLY,
_ => Rotation::DAILY,
};
let mut builder = RollingFileAppender::builder()
.rotation(rotation)
.max_log_files(keep_files * 3); // Make sure there are some data files to archive to avoid premature deletion
match match_mode {
FileMatchMode::Prefix => builder = builder.filename_prefix(log_filename),
FileMatchMode::Suffix => builder = builder.filename_suffix(log_filename),
}
builder
.build(log_directory)
.map_err(|e| TelemetryError::Io(format!("failed to initialize rolling file appender: {e}")))?
let rotation = match rotation_str.as_str() {
"minutely" => Rotation::Minutely,
"hourly" => Rotation::Hourly,
"daily" => Rotation::Daily,
_ => Rotation::Daily,
};
let max_single_file_size = config
.log_max_single_file_size_bytes
.unwrap_or(DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES);
let file_appender =
RollingAppender::new(log_directory, log_filename.to_string(), rotation, max_single_file_size, match_mode)?;
let (non_blocking, guard) = tracing_appender::non_blocking(file_appender);
// ── 4. Build subscriber layers ────────────────────────────────────────────
@@ -270,7 +263,7 @@ fn init_file_logging_internal(
info!(
"Init file logging at '{}', rotation: {}, keep {} files",
log_directory, rotation, keep_files
log_directory, rotation_str, keep_files
);
Ok(OtelGuard {
@@ -352,6 +345,7 @@ fn spawn_cleanup_task(
// Use suffix matching for log files like "2026-03-01-06-21.rustfs.log"
// where "rustfs.log" is the suffix.
let file_pattern = config.log_filename.as_deref().unwrap_or(log_filename).to_string();
let active_filename = file_pattern.clone();
// Determine match mode from config, defaulting to Suffix
let match_mode = match config.log_match_mode.as_deref().map(|s| s.to_lowercase()).as_deref() {
@@ -386,21 +380,21 @@ fn spawn_cleanup_task(
.log_cleanup_interval_seconds
.unwrap_or(DEFAULT_OBS_LOG_CLEANUP_INTERVAL_SECONDS);
let cleaner = Arc::new(LogCleaner::new(
log_dir,
file_pattern,
match_mode,
keep_files,
max_total_size,
max_single_file_size,
compress,
gzip_level,
retention_days,
exclude_patterns,
delete_empty,
min_age,
dry_run,
));
let cleaner = Arc::new(
LogCleaner::builder(log_dir, file_pattern, active_filename)
.match_mode(match_mode)
.keep_files(keep_files)
.max_total_size_bytes(max_total_size)
.max_single_file_size_bytes(max_single_file_size)
.compress_old_files(compress)
.gzip_compression_level(gzip_level)
.compressed_file_retention_days(retention_days)
.exclude_patterns(exclude_patterns)
.delete_empty_files(delete_empty)
.min_file_age_seconds(min_age)
.dry_run(dry_run)
.build(),
);
tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(cleanup_interval));
@@ -433,8 +427,13 @@ mod tests {
..OtelConfig::default()
};
let result = init_file_logging_internal(&config, temp_path, "info", true);
assert!(result.is_err());
// We must run within a Tokio runtime because init_file_logging_internal spawns a background task.
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let result = init_file_logging_internal(&config, temp_path, "info", true);
// With eager file opening, an invalid filename (null byte) causes the OS to reject
// the open() call, so the function returns Err instead of panicking.
assert!(result.is_err(), "invalid filename must return Err, not panic");
});
}
}
crates/obs/src/telemetry/mod.rs
+9 -4
View File
@@ -45,6 +45,7 @@ mod local;
mod otel;
mod recorder;
mod resource;
mod rolling;
use crate::TelemetryError;
use crate::config::OtelConfig;
@@ -117,8 +118,9 @@ pub(crate) fn init_telemetry(config: &OtelConfig) -> Result<OtelGuard, Telemetry
#[cfg(test)]
mod tests {
use super::*;
use rustfs_config::observability::DEFAULT_OBS_ENVIRONMENT_PRODUCTION;
use rustfs_config::{ENVIRONMENT, USE_STDOUT};
use rustfs_config::{DEFAULT_OBS_LOG_STDOUT_ENABLED, ENVIRONMENT};
#[test]
fn test_production_environment_detection() {
@@ -160,7 +162,7 @@ mod tests {
TestCase {
is_production: false,
config_use_stdout: None,
expected_use_stdout: USE_STDOUT,
expected_use_stdout: DEFAULT_OBS_LOG_STDOUT_ENABLED,
description: "Non-production with no config should use default",
},
TestCase {
@@ -184,7 +186,11 @@ mod tests {
];
for case in &test_cases {
let default_use_stdout = if case.is_production { false } else { USE_STDOUT };
let default_use_stdout = if case.is_production {
false
} else {
DEFAULT_OBS_LOG_STDOUT_ENABLED
};
let actual = case.config_use_stdout.unwrap_or(default_use_stdout);
assert_eq!(actual, case.expected_use_stdout, "Test case failed: {}", case.description);
}
@@ -221,7 +227,6 @@ mod tests {
#[test]
fn test_otel_config_environment_defaults() {
// Verify that environment field defaults behave correctly.
use crate::config::OtelConfig;
let config = OtelConfig {
endpoint: "".to_string(),
use_stdout: None,
crates/obs/src/telemetry/rolling.rs
+508
View File
@@ -0,0 +1,508 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! A custom rolling file appender that supports both time-based and size-based rotation.
//!
//! This is a lightweight replacement for `tracing_appender::rolling::RollingFileAppender`
//! which only supports time-based rotation. This implementation ensures that active
//! log files do not grow indefinitely by rotating them when they exceed a configured size.
use crate::cleaner::types::FileMatchMode;
use jiff::Zoned;
use std::fs::{self, File};
use std::io::{self, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::thread;
use std::time::Duration;
#[derive(Debug, Clone, Copy)]
pub enum Rotation {
Minutely,
Hourly,
Daily,
#[allow(dead_code)]
Never,
}
/// Global per-process counter used to disambiguate archive filenames that
/// may otherwise collide when multiple rotations occur within the same
/// timestamp tick.
static ROLL_UNIQUIFIER: AtomicU64 = AtomicU64::new(0);
impl Rotation {
fn check_should_roll(&self, last: i64, now: i64) -> bool {
match self {
Rotation::Minutely => now / 60 != last / 60,
Rotation::Hourly => now / 3600 != last / 3600,
Rotation::Daily => {
// Align daily rotation with the local day boundary rather than UTC midnight.
// We shift both timestamps by the current local offset before bucketing into days.
let offset_secs = Zoned::now().offset().seconds() as i64;
(now + offset_secs) / 86400 != (last + offset_secs) / 86400
}
Rotation::Never => false,
}
}
}
pub struct RollingAppender {
dir: PathBuf,
filename: String,
rotation: Rotation,
max_size_bytes: u64,
match_mode: FileMatchMode,
file: Option<File>,
size: u64,
// Store as seconds since Unix epoch
last_roll_ts: i64,
}
impl RollingAppender {
/// Create and immediately validate a new `RollingAppender`.
///
/// The log directory is created if it does not already exist, and the
/// active log file is opened (or created) eagerly so that configuration
/// errors — e.g. an invalid filename — surface at initialisation time
/// rather than on the first write.
///
/// # Errors
/// Returns an [`io::Error`] if:
/// - `filename` is not a plain file name (absolute path, path separators,
/// or `..` components are rejected to prevent path traversal).
/// - The directory cannot be created.
/// - The active log file cannot be opened/created.
pub fn new(
dir: impl AsRef<Path>,
filename: String,
rotation: Rotation,
max_size_bytes: u64,
match_mode: FileMatchMode,
) -> io::Result<Self> {
// Validate that `filename` is a plain file name: not absolute and no
// directory components (separators or `..`). If `file_name()` equals
// the entire path, there can be no parent-directory traversal.
{
let p = Path::new(&filename);
let is_plain_name = !p.is_absolute() && p.file_name().map(|n| n == p.as_os_str()).unwrap_or(false);
if !is_plain_name {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("log filename must be a plain file name with no path components, got: {filename:?}"),
));
}
}
let mut appender = Self {
dir: dir.as_ref().to_path_buf(),
filename,
rotation,
max_size_bytes,
match_mode,
file: None,
size: 0,
last_roll_ts: Zoned::now().timestamp().as_second(),
};
// Eagerly open the file to validate the path and capture accurate
// initial size / last-roll timestamp.
appender.open_file()?;
Ok(appender)
}
fn active_file_path(&self) -> PathBuf {
self.dir.join(&self.filename)
}
fn open_file(&mut self) -> io::Result<()> {
if self.file.is_some() {
return Ok(());
}
let path = self.active_file_path();
// Ensure directory exists
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)?;
}
// Open in append mode
let file = fs::OpenOptions::new().create(true).append(true).open(&path)?;
let meta = file.metadata()?;
self.size = meta.len();
// Seed `last_roll_ts` from the file's modification time so that a
// process restart correctly triggers time-based rotation if the active
// file belongs to a previous period.
if let Ok(modified) = meta.modified() {
// Convert SystemTime to jiff::Timestamp
if let Ok(ts) = jiff::Timestamp::try_from(modified) {
self.last_roll_ts = ts.as_second();
}
}
self.file = Some(file);
Ok(())
}
fn should_roll(&self, write_len: u64) -> bool {
// 1. Size-based check (Cheap, check first)
// If max_size is set (non-zero) and writing would exceed it, roll immediately.
if self.max_size_bytes > 0 && (self.size + write_len) > self.max_size_bytes {
return true;
}
// 2. Time-based check
// We check this after size check to avoid unnecessary time calls if size forces a roll.
let now = Zoned::now().timestamp().as_second();
self.rotation.check_should_roll(self.last_roll_ts, now)
}
fn roll(&mut self) -> io::Result<()> {
// 1. Close current file first to ensure all buffers are flushed to OS (if any)
// and handle released.
self.file = None;
let active_path = self.active_file_path();
if !active_path.exists() {
return Ok(());
}
// 2. Generate archive name.
// Format: YYYYMMDDHHMMSS.uuuuuu (Microsecond/Nanosecond precision)
// We use jiff's strftime. "%Y%m%d%H%M%S%.6f" gives microsecond precision.
let now = Zoned::now();
let timestamp_str = now.strftime("%Y%m%d%H%M%S%.6f").to_string();
// Add a unique counter to prevent collisions in high-concurrency/fast-rotation scenarios.
let counter = ROLL_UNIQUIFIER.fetch_add(1, Ordering::Relaxed);
// Final suffix/prefix part: timestamp + counter
// Example: 20231027103001.123456-0
let unique_part = format!("{}-{}", timestamp_str, counter);
// Match naming strategy with LogCleaner expectations.
let archive_name = match self.match_mode {
FileMatchMode::Suffix => {
// Suffix mode: timestamp BEFORE filename.
// e.g. rustfs.log -> 20231027103001.123456-0.rustfs.log
format!("{}.{}", unique_part, self.filename)
}
FileMatchMode::Prefix => {
// Prefix mode: timestamp AFTER filename.
// e.g. rustfs -> rustfs.20231027103001.123456-0
format!("{}.{}", self.filename, unique_part)
}
};
// 3. Rename the active file to the archive path.
let archive_path = self.dir.join(&archive_name);
// Robust Rename Strategy:
// On Windows, file locking (e.g. by AV software or indexers) can cause `rename` to fail
// spuriously with PermissionDenied. We implement a short retry loop with backoff.
const MAX_RETRIES: u32 = 3;
let mut last_error = None;
for i in 0..MAX_RETRIES {
match fs::rename(&active_path, &archive_path) {
Ok(_) => {
// Success!
// 4. Reset state
self.size = 0;
self.last_roll_ts = now.timestamp().as_second();
// 5. Re-open (creates new active file)
self.open_file()?;
return Ok(());
}
Err(e) => {
// Decide if we should retry based on error kind
let should_retry = match e.kind() {
// Windows often returns PermissionDenied for locked files
io::ErrorKind::PermissionDenied => true,
io::ErrorKind::Interrupted => true,
_ => false,
};
last_error = Some(e);
if !should_retry {
break;
}
// Exponential backoff: 10ms, 20ms, 40ms...
thread::sleep(Duration::from_millis(10 * (1 << i)));
}
}
}
// 6. Recovery Failure
// If we exhausted retries, we MUST NOT lose log data.
// We re-open the ACTIVE file (which is still there because rename failed).
// The file will grow beyond max_size, but availability > strict sizing.
eprintln!(
"RollingAppender: Failed to rotate log file after {} retries. Error: {:?}",
MAX_RETRIES, last_error
);
// Attempt to re-open existing active file to allow continued writing
self.open_file()?;
// Return the error so it can be logged/handled, even though we recovered the handle.
Err(last_error.unwrap_or_else(|| io::Error::other("Unknown rename error")))
}
}
impl Write for RollingAppender {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
// Ensure file is open
if self.file.is_none() {
self.open_file()?;
}
// Check rotation
if self.should_roll(buf.len() as u64)
&& let Err(e) = self.roll()
{
// If rotation fails, we log to stderr and try to continue writing to the active file
// to avoid losing logs if possible.
eprintln!("RollingAppender: failed to rotate log file: {}", e);
}
// Ensure file is open (in case roll closed it and failed to open new one, or open_file failed above)
if self.file.is_none() {
self.open_file()?;
}
if let Some(file) = &mut self.file {
let n = file.write(buf)?;
self.size += n as u64;
Ok(n)
} else {
Err(io::Error::other("Failed to open log file"))
}
}
fn flush(&mut self) -> io::Result<()> {
if let Some(file) = &mut self.file {
file.flush()
} else {
Ok(())
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::fs;
use tempfile::TempDir;
fn count_files(dir: &Path) -> usize {
fs::read_dir(dir)
.unwrap()
.filter(|e| e.as_ref().unwrap().path().is_file())
.count()
}
// ── Construction ──────────────────────────────────────────────────────────
#[test]
fn test_new_creates_file_eagerly() {
let tmp = TempDir::new().unwrap();
let _appender = RollingAppender::new(tmp.path(), "test.log".to_string(), Rotation::Daily, 0, FileMatchMode::Suffix)
.expect("should create appender without error");
assert!(tmp.path().join("test.log").exists(), "active log file should be created on new()");
}
#[test]
fn test_new_invalid_filename_returns_error() {
let tmp = TempDir::new().unwrap();
// Null byte is invalid on both Unix and Windows.
let result = RollingAppender::new(tmp.path(), "invalid\0name.log".to_string(), Rotation::Daily, 0, FileMatchMode::Suffix);
assert!(result.is_err(), "null byte in filename must produce an error");
}
#[test]
fn test_new_rejects_path_with_separators() {
let tmp = TempDir::new().unwrap();
// A filename containing path separators could escape the log directory.
let result = RollingAppender::new(tmp.path(), "subdir/app.log".to_string(), Rotation::Daily, 0, FileMatchMode::Suffix);
assert!(result.is_err(), "filename with path separator must be rejected");
}
#[test]
fn test_new_rejects_parent_directory_traversal() {
let tmp = TempDir::new().unwrap();
// "../secret.log" would write outside the log directory.
let result = RollingAppender::new(tmp.path(), "../secret.log".to_string(), Rotation::Daily, 0, FileMatchMode::Suffix);
assert!(result.is_err(), "parent-directory traversal in filename must be rejected");
}
#[test]
fn test_new_rejects_absolute_path_as_filename() {
let tmp = TempDir::new().unwrap();
let result = RollingAppender::new(tmp.path(), "/etc/app.log".to_string(), Rotation::Daily, 0, FileMatchMode::Suffix);
assert!(result.is_err(), "absolute path as filename must be rejected");
}
/// On Windows, backslash is a path separator and must be rejected.
#[cfg(windows)]
#[test]
fn test_new_rejects_backslash_path_separator_on_windows() {
let tmp = TempDir::new().unwrap();
let result = RollingAppender::new(tmp.path(), "subdir\\app.log".to_string(), Rotation::Daily, 0, FileMatchMode::Suffix);
assert!(result.is_err(), "backslash path separator in filename must be rejected on Windows");
}
// ── Basic writes ──────────────────────────────────────────────────────────
#[test]
fn test_write_stores_content() {
let tmp = TempDir::new().unwrap();
let mut appender =
RollingAppender::new(tmp.path(), "test.log".to_string(), Rotation::Daily, 0, FileMatchMode::Suffix).unwrap();
appender.write_all(b"hello world\n").expect("write should succeed");
appender.flush().expect("flush should succeed");
let content = fs::read_to_string(tmp.path().join("test.log")).unwrap();
assert_eq!(content, "hello world\n");
}
// ── Size-based rotation ────────────────────────────────────────────────────
#[test]
fn test_size_rotation_creates_archive() {
let tmp = TempDir::new().unwrap();
// Allow only 5 bytes before rotating.
let mut appender =
RollingAppender::new(tmp.path(), "app.log".to_string(), Rotation::Never, 5, FileMatchMode::Suffix).unwrap();
// First write: 5 bytes exactly — no rotation yet.
appender.write_all(b"12345").expect("write should succeed");
// Second write: would push past the limit — rotation should occur first.
appender.write_all(b"abcde").expect("write after rotation should succeed");
appender.flush().unwrap();
// There should now be 2 files: the active log + 1 archive.
assert_eq!(count_files(tmp.path()), 2, "one rotation should have produced one archive");
// The active file should only contain the second write.
let content = fs::read_to_string(tmp.path().join("app.log")).unwrap();
assert_eq!(content, "abcde");
}
#[test]
fn test_multiple_size_rotations_produce_unique_archives() {
let tmp = TempDir::new().unwrap();
// Force a rotation on every write of 4+ bytes.
let mut appender =
RollingAppender::new(tmp.path(), "app.log".to_string(), Rotation::Never, 3, FileMatchMode::Suffix).unwrap();
for _ in 0..5 {
appender.write_all(b"abcd").expect("write should succeed");
}
appender.flush().unwrap();
let file_count = count_files(tmp.path());
// At least 5 archives (one per rotation) plus the active file.
assert!(
file_count >= 5,
"each burst write should produce a distinct archive; got {file_count} files"
);
}
// ── Archive filename format ────────────────────────────────────────────────
#[test]
fn test_suffix_mode_archive_name() {
let tmp = TempDir::new().unwrap();
let mut appender =
RollingAppender::new(tmp.path(), "app.log".to_string(), Rotation::Never, 3, FileMatchMode::Suffix).unwrap();
appender.write_all(b"1234").expect("write should succeed");
appender.flush().unwrap();
let archives: Vec<_> = fs::read_dir(tmp.path())
.unwrap()
.filter_map(|e| e.ok())
.map(|e| e.file_name().to_string_lossy().to_string())
.filter(|n| n != "app.log")
.collect();
assert_eq!(archives.len(), 1);
// Suffix mode: "<timestamp>-<counter>.app.log"
// Since timestamp contains digits and we use high precision, checking strictly is hard,
// but it should definitely NOT be the old unix timestamp format (just digits).
// It should contain "-" before "app.log" due to our new format.
assert!(
archives[0].ends_with(".app.log"),
"archive should end with '.app.log' in Suffix mode; got '{}'",
archives[0]
);
// Check for new format chars (YMD)
// 20xx...
assert!(
archives[0].starts_with("20"),
"archive should start with year (20xx); got '{}'",
archives[0]
);
}
#[test]
fn test_prefix_mode_archive_name() {
let tmp = TempDir::new().unwrap();
let mut appender =
RollingAppender::new(tmp.path(), "app".to_string(), Rotation::Never, 3, FileMatchMode::Prefix).unwrap();
appender.write_all(b"1234").expect("write should succeed");
appender.flush().unwrap();
let archives: Vec<_> = fs::read_dir(tmp.path())
.unwrap()
.filter_map(|e| e.ok())
.map(|e| e.file_name().to_string_lossy().to_string())
.filter(|n| n != "app")
.collect();
assert_eq!(archives.len(), 1);
// Prefix mode: "app.<timestamp>-<counter>"
assert!(
archives[0].starts_with("app.20"),
"archive should start with 'app.20' in Prefix mode; got '{}'",
archives[0]
);
}
// ── Restart with existing file ─────────────────────────────────────────────
#[test]
fn test_restart_with_existing_file_reads_size() {
let tmp = TempDir::new().unwrap();
let log_path = tmp.path().join("app.log");
fs::write(&log_path, b"existing content").unwrap();
let appender =
RollingAppender::new(tmp.path(), "app.log".to_string(), Rotation::Daily, 0, FileMatchMode::Suffix).unwrap();
assert_eq!(
appender.size,
b"existing content".len() as u64,
"size should reflect existing file content"
);
}
}