diff --git a/crates/madmin/src/service_commands.rs b/crates/madmin/src/service_commands.rs index d5ee29e85..ddda978be 100644 --- a/crates/madmin/src/service_commands.rs +++ b/crates/madmin/src/service_commands.rs @@ -93,13 +93,6 @@ impl ServiceTraceOpts { self.batch_replication = query_pairs.get("batch-replication").is_some_and(|v| v == "true"); self.batch_key_rotation = query_pairs.get("batch-keyrotation").is_some_and(|v| v == "true"); self.batch_expire = query_pairs.get("batch-expire").is_some_and(|v| v == "true"); - if query_pairs.get("all").is_some_and(|v| v == "true") { - self.s3 = true; - self.internal = true; - self.storage = true; - self.os = true; - } - self.rebalance = query_pairs.get("rebalance").is_some_and(|v| v == "true"); self.storage = query_pairs.get("storage").is_some_and(|v| v == "true"); self.internal = query_pairs.get("internal").is_some_and(|v| v == "true"); @@ -109,6 +102,13 @@ impl ServiceTraceOpts { self.ftp = query_pairs.get("ftp").is_some_and(|v| v == "true"); self.ilm = query_pairs.get("ilm").is_some_and(|v| v == "true"); + if query_pairs.get("all").is_some_and(|v| v == "true") { + self.s3 = true; + self.internal = true; + self.storage = true; + self.os = true; + } + if let Some(threshold) = query_pairs.get("threshold") { let duration = parse_duration(threshold)?; self.threshold = duration; @@ -117,3 +117,22 @@ impl ServiceTraceOpts { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_params_all_true_keeps_internal_and_storage_enabled() { + let uri: Uri = "/trace?all=true".parse().expect("valid uri"); + let mut opts = ServiceTraceOpts::default(); + + opts.parse_params(&uri).expect("all=true should parse"); + let trace_types = opts.trace_types(); + + assert!(trace_types.contains(&TraceType::S3)); + assert!(trace_types.contains(&TraceType::INTERNAL)); + assert!(trace_types.contains(&TraceType::STORAGE)); + assert!(trace_types.contains(&TraceType::OS)); + } +} diff --git a/crates/obs/src/cleaner/README.md b/crates/obs/src/cleaner/README.md index dfadd6d72..aa0af20a4 100644 --- a/crates/obs/src/cleaner/README.md +++ b/crates/obs/src/cleaner/README.md @@ -73,7 +73,7 @@ These values can be wired into dashboards and alert rules for cleanup health. For regular logs, the cleaner evaluates candidates in this order: -1. keep at least `keep_files` newest matching generations; +1. keep at most `keep_files` newest matching generations; 2. remove older files if total retained size still exceeds `max_total_size_bytes`; 3. remove any file whose individual size exceeds `max_single_file_size_bytes`; 4. if compression is enabled, archive before deletion; diff --git a/crates/obs/src/cleaner/core.rs b/crates/obs/src/cleaner/core.rs index fefaf6732..ae9d906eb 100644 --- a/crates/obs/src/cleaner/core.rs +++ b/crates/obs/src/cleaner/core.rs @@ -69,7 +69,7 @@ pub struct LogCleaner { pub(super) active_filename: String, /// Whether `file_pattern` is interpreted as a prefix or suffix. pub(super) match_mode: FileMatchMode, - /// Minimum number of regular log files to keep regardless of size. + /// Maximum number of newest regular log files to keep. pub(super) keep_files: usize, /// Optional cap for the cumulative size of regular logs. pub(super) max_total_size_bytes: u64, @@ -205,8 +205,8 @@ impl LogCleaner { /// Choose regular log files that should be compressed and/or deleted. /// /// The `files` slice must already be sorted from oldest to newest. The - /// method first preserves the newest `keep_files` generations, then applies - /// total-size and per-file-size limits to the remaining tail. + /// method first enforces the `keep_files` ceiling, then applies total-size + /// and per-file-size limits to the remaining tail. pub(super) fn select_files_to_process(&self, files: &[FileInfo], total_size: u64) -> Vec { let mut to_delete = Vec::new(); if files.is_empty() { @@ -325,8 +325,6 @@ impl LogCleaner { } else { match injector.steal_batch_and_pop(&local_worker) { Steal::Success(file) => { - attempts.fetch_add(1, Ordering::Relaxed); - successes.fetch_add(1, Ordering::Relaxed); Some(file) } Steal::Retry => continue, @@ -357,6 +355,7 @@ impl LogCleaner { continue; }; + let mut file = file; let compressed = match compress_file(&file.path, &options) { Ok(output) => { debug!( @@ -371,6 +370,7 @@ impl LogCleaner { state = "parallel_compression_done", "log cleaner state changed" ); + file.projected_freed_bytes = output.input_bytes.saturating_sub(output.output_bytes); true } Err(err) => { @@ -508,7 +508,9 @@ impl LogCleaner { state = "serial_compression_done", "log cleaner state changed" ); - deletable.push(file.clone()); + let mut file = file.clone(); + file.projected_freed_bytes = output.input_bytes.saturating_sub(output.output_bytes); + deletable.push(file); } Err(err) => { warn!(event = EVENT_LOG_CLEANER_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_LOG_CLEANER, file = ?file.path, error = %err, result = "serial_compression_failed", "log cleaner state changed"); @@ -585,15 +587,15 @@ impl LogCleaner { if self.dry_run { info!(event = EVENT_LOG_CLEANER_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_LOG_CLEANER, state = "dry_run_delete", file = ?f.path, bytes = f.size, "log cleaner state changed"); deleted += 1; - freed += f.size; + freed += f.projected_freed_bytes; continue; } match self.secure_delete(&f.path) { Ok(()) => { deleted += 1; - freed += f.size; - debug!(event = EVENT_LOG_CLEANER_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_LOG_CLEANER, state = "deleted", file = ?f.path, bytes = f.size, "log cleaner state changed"); + freed += f.projected_freed_bytes; + debug!(event = EVENT_LOG_CLEANER_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_LOG_CLEANER, state = "deleted", file = ?f.path, bytes = f.size, projected_freed_bytes = f.projected_freed_bytes, "log cleaner state changed"); } Err(e) => { error!(event = EVENT_LOG_CLEANER_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_LOG_CLEANER, result = "delete_failed", file = ?f.path, error = %e, "log cleaner state changed"); @@ -665,7 +667,7 @@ impl LogCleanerBuilder { self } - /// Preserve at least this many newest regular log files. + /// Keep at most this many newest regular log files. pub fn keep_files(mut self, keep_files: usize) -> Self { self.keep_files = keep_files; self diff --git a/crates/obs/src/cleaner/mod.rs b/crates/obs/src/cleaner/mod.rs index a3cef0d63..2fe7465f1 100644 --- a/crates/obs/src/cleaner/mod.rs +++ b/crates/obs/src/cleaner/mod.rs @@ -127,8 +127,6 @@ mod tests { fn assert_parallel_cleanup_completes(file_count: usize, workers: usize) -> std::io::Result<()> { let tmp = TempDir::new()?; let dir = tmp.path().to_path_buf(); - let expected_freed = (file_count * 256) as u64; - for i in 0..file_count { create_log_file(&dir, &format!("app.log.2024-01-{i:03}"), 256)?; } @@ -160,9 +158,22 @@ mod tests { name.starts_with("app.log.") && !compressed_suffixes.iter().any(|suffix| name.ends_with(suffix)) }) .count(); + let archive_bytes: u64 = std::fs::read_dir(&dir)? + .filter_map(Result::ok) + .filter(|entry| { + let name = entry.file_name(); + let name = name.to_string_lossy(); + compressed_suffixes.iter().any(|suffix| name.ends_with(suffix)) + }) + .map(|entry| entry.metadata().map(|metadata| metadata.len()).unwrap_or(0)) + .sum(); assert_eq!(result.0, file_count, "all rotated logs should be deleted after compression"); - assert_eq!(result.1, expected_freed, "freed bytes should match the removed source files"); + assert_eq!( + result.1, + (file_count * 256) as u64 - archive_bytes, + "freed bytes should exclude archive bytes that still remain on disk" + ); assert_eq!(archive_count, file_count, "each rotated log should leave behind one archive"); assert_eq!(original_count, 0, "compressed source logs should be removed"); Ok(()) diff --git a/crates/obs/src/cleaner/scanner.rs b/crates/obs/src/cleaner/scanner.rs index b48c1de40..458cc2b0c 100644 --- a/crates/obs/src/cleaner/scanner.rs +++ b/crates/obs/src/cleaner/scanner.rs @@ -199,6 +199,7 @@ pub(super) fn scan_log_directory( let info = FileInfo { path, size: file_size, + projected_freed_bytes: file_size, modified, }; diff --git a/crates/obs/src/cleaner/types.rs b/crates/obs/src/cleaner/types.rs index e0fc332e6..201589aaa 100644 --- a/crates/obs/src/cleaner/types.rs +++ b/crates/obs/src/cleaner/types.rs @@ -176,6 +176,11 @@ pub(super) struct FileInfo { /// /// This value is used for retention accounting and freed-byte metrics. pub size: u64, + /// Projected bytes reclaimed when this file is deleted. + /// + /// For source logs that were compressed first, this reflects the original + /// size minus the archive bytes that remain on disk. + pub projected_freed_bytes: u64, /// Last-modification timestamp from the filesystem. /// /// The selection phase sorts on this timestamp so the oldest files are diff --git a/crates/obs/src/config.rs b/crates/obs/src/config.rs index 8c86087f4..a9b531786 100644 --- a/crates/obs/src/config.rs +++ b/crates/obs/src/config.rs @@ -80,7 +80,7 @@ const LEGACY_ENV_OBS_PROFILING_ENABLED: &str = "RUSTFS_OBS_PROFILING_ENABLED"; /// - All fields are `Option` to allow partial configuration via environment /// variables with sensible defaults provided by constants in `rustfs-config`. /// - `log_keep_files` is used to derive the rolling-appender's upper bound on -/// retained files (if enabled) and to set the cleaner's minimum retention count. +/// retained files (if enabled) and to set the cleaner's retained-file ceiling. /// /// # Example /// ```no_run diff --git a/crates/obs/src/metrics/scheduler.rs b/crates/obs/src/metrics/scheduler.rs index 475627110..a6844eec4 100644 --- a/crates/obs/src/metrics/scheduler.rs +++ b/crates/obs/src/metrics/scheduler.rs @@ -403,7 +403,7 @@ fn parse_metrics_interval(primary_env: &str, legacy_env: &str, default_interval: fn parse_system_metrics_interval() -> Duration { get_env_opt_u64(ENV_SYSTEM_METRICS_INTERVAL) - .or_else(|| get_env_opt_u64(LEGACY_SYSTEM_METRICS_INTERVAL).map(|ms| ms / 1000)) + .or_else(|| get_env_opt_u64(LEGACY_SYSTEM_METRICS_INTERVAL).map(|ms| if ms == 0 { 0 } else { ms.div_ceil(1000) })) .or_else(|| get_env_opt_u64(ENV_DEFAULT_METRICS_INTERVAL)) .filter(|&v| v > 0) .map(Duration::from_secs) @@ -1280,6 +1280,10 @@ pub fn init_metrics_runtime(token: CancellationToken) { let mut next_resource_run = now; let mut next_system_run = now; + host_system.refresh_cpu_all(); + tokio::time::sleep(sysinfo::MINIMUM_CPU_UPDATE_INTERVAL).await; + host_system.refresh_cpu_all(); + #[cfg(feature = "gpu")] let current_pid = match sysinfo::get_current_pid() { Ok(pid) => Some(pid), @@ -1800,4 +1804,32 @@ mod tests { expire_series_zero_tombstones(&mut zero_tombstones); assert!(zero_tombstones.is_empty()); } + + #[test] + fn parse_system_metrics_interval_rounds_legacy_millis_up_to_one_second() { + temp_env::with_vars( + [ + (ENV_SYSTEM_METRICS_INTERVAL, None::<&str>), + (LEGACY_SYSTEM_METRICS_INTERVAL, Some("500")), + (ENV_DEFAULT_METRICS_INTERVAL, None::<&str>), + ], + || { + assert_eq!(parse_system_metrics_interval(), Duration::from_secs(1)); + }, + ); + } + + #[test] + fn parse_system_metrics_interval_rounds_legacy_millis_up() { + temp_env::with_vars( + [ + (ENV_SYSTEM_METRICS_INTERVAL, None::<&str>), + (LEGACY_SYSTEM_METRICS_INTERVAL, Some("1500")), + (ENV_DEFAULT_METRICS_INTERVAL, None::<&str>), + ], + || { + assert_eq!(parse_system_metrics_interval(), Duration::from_secs(2)); + }, + ); + } } diff --git a/crates/obs/src/telemetry/filter.rs b/crates/obs/src/telemetry/filter.rs index 999fca951..d0e9c1df7 100644 --- a/crates/obs/src/telemetry/filter.rs +++ b/crates/obs/src/telemetry/filter.rs @@ -150,6 +150,16 @@ fn should_demote_http_request_logs(logger_level: &str, default_level: Option<&st matches!(level.as_str(), "info" | "warn") } +fn rust_log_explicitly_names_target(rust_log: &str, target: &str) -> bool { + rust_log.split(',').map(str::trim).any(|directive| { + if let Some((directive_target, _)) = directive.rsplit_once('=') { + directive_target.trim() == target + } else { + false + } + }) +} + pub(super) fn build_env_filter(logger_level: &str, default_level: Option<&str>) -> EnvFilter { // 1. Determine the base filter source. // If `default_level` is set (e.g. forced override), we use it. @@ -189,6 +199,13 @@ pub(super) fn build_env_filter(logger_level: &str, default_level: Option<&str>) } for (crate_name, level) in directives { + if rust_log_env + .as_deref() + .is_some_and(|rust_log| rust_log_explicitly_names_target(rust_log, crate_name)) + { + continue; + } + // We use `add_directive` which effectively appends to the filter. // If RUST_LOG already specified `hyper=debug`, adding `hyper=off` later MIGHT override it // depending on specificity, but usually the last directive wins or the most specific one. @@ -377,4 +394,17 @@ mod tests { ); }); } + + #[test] + fn test_build_env_filter_preserves_explicit_target_directive() { + temp_env::with_var("RUST_LOG", Some("info,hyper=info"), || { + let filter = build_env_filter("info", None); + let filter_str = filter.to_string().to_ascii_lowercase(); + + assert!( + !filter_str.contains("hyper=off"), + "suppression must not override an explicit target directive: {filter_str}" + ); + }); + } } diff --git a/crates/obs/src/telemetry/recorder.rs b/crates/obs/src/telemetry/recorder.rs index 9c0b3e30e..1b49642d2 100644 --- a/crates/obs/src/telemetry/recorder.rs +++ b/crates/obs/src/telemetry/recorder.rs @@ -111,7 +111,7 @@ impl Builder { } } -#[derive(Debug)] +#[derive(Debug, Clone)] struct MetricMetadata { unit: Option, description: SharedString, @@ -196,7 +196,7 @@ impl Recorder { } fn get_metadata_for_builder(&self, key_name: &str) -> Option { - self.with_metadata_lock(|metadata| metadata.remove(key_name)) + self.with_metadata_lock(|metadata| metadata.get(key_name).cloned()) } } @@ -264,6 +264,7 @@ impl metrics::Recorder for Recorder { gauge, labels, value: AtomicU64::new(0), + record_lock: Mutex::new(()), })); Self::insert_cached_metric(&self.cached_gauges, key.clone(), handle, "gauge") @@ -313,10 +314,12 @@ struct WrappedGauge { gauge: opentelemetry::metrics::Gauge, labels: Vec, value: AtomicU64, + record_lock: Mutex<()>, } impl GaugeFn for WrappedGauge { fn increment(&self, value: f64) { + let _guard = self.record_lock.lock().unwrap_or_else(|e| e.into_inner()); let mut current = self.value.load(Ordering::Relaxed); let mut new = f64::from_bits(current) + value; while let Err(val) = self @@ -331,6 +334,7 @@ impl GaugeFn for WrappedGauge { } fn decrement(&self, value: f64) { + let _guard = self.record_lock.lock().unwrap_or_else(|e| e.into_inner()); let mut current = self.value.load(Ordering::Relaxed); let mut new = f64::from_bits(current) - value; while let Err(val) = self @@ -345,6 +349,7 @@ impl GaugeFn for WrappedGauge { } fn set(&self, value: f64) { + let _guard = self.record_lock.lock().unwrap_or_else(|e| e.into_inner()); self.value.store(value.to_bits(), Ordering::Relaxed); self.gauge.record(value, &self.labels); } @@ -473,4 +478,25 @@ mod tests { let cache = shared.cached_counters.read().unwrap(); assert_eq!(cache.len(), 1, "concurrent registrations should produce exactly one cache entry"); } + + #[test] + fn metadata_is_available_for_multiple_label_variants() { + let recorder = test_recorder(); + recorder.describe_counter("shared_counter".into(), Unit::from_string("bytes"), "shared description".into()); + + let first = Key::from_parts("shared_counter", vec![metrics::Label::new("kind", "a")]); + let second = Key::from_parts("shared_counter", vec![metrics::Label::new("kind", "b")]); + let meta = test_metadata(); + + let _ = recorder.register_counter(&first, &meta); + let second_metadata = recorder.get_metadata_for_builder("shared_counter"); + assert!(second_metadata.is_some()); + assert_eq!(second_metadata.as_ref().and_then(|metadata| metadata.unit), Unit::from_string("bytes")); + assert_eq!( + second_metadata.as_ref().map(|metadata| metadata.description.to_string()), + Some("shared description".to_string()) + ); + + let _ = recorder.register_counter(&second, &meta); + } } diff --git a/crates/obs/src/telemetry/rolling.rs b/crates/obs/src/telemetry/rolling.rs index 45f297e45..d2471e180 100644 --- a/crates/obs/src/telemetry/rolling.rs +++ b/crates/obs/src/telemetry/rolling.rs @@ -23,6 +23,7 @@ use crate::global::{ METRIC_LOG_CLEANER_ACTIVE_FILE_SIZE_BYTES, METRIC_LOG_CLEANER_ROTATION_DURATION_SECONDS, METRIC_LOG_CLEANER_ROTATION_FAILURES_TOTAL, METRIC_LOG_CLEANER_ROTATION_TOTAL, }; +use chrono::{Local, TimeZone as _}; use jiff::Zoned; use metrics::{counter, gauge, histogram}; use std::fs::{self, File}; @@ -30,7 +31,7 @@ use std::io::{self, Write}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; use std::thread; -use std::time::Duration; +use std::time::{Duration, Instant}; #[derive(Debug, Clone, Copy)] pub enum Rotation { @@ -45,6 +46,7 @@ pub enum Rotation { /// may otherwise collide when multiple rotations occur within the same /// timestamp tick. static ROLL_UNIQUIFIER: AtomicU64 = AtomicU64::new(0); +const ROTATION_RETRY_COOLDOWN: Duration = Duration::from_secs(5); impl Rotation { fn check_should_roll(&self, last: i64, now: i64) -> bool { @@ -52,10 +54,13 @@ impl Rotation { 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 + let last_day = Local.timestamp_opt(last, 0).single().map(|ts| ts.date_naive()); + let now_day = Local.timestamp_opt(now, 0).single().map(|ts| ts.date_naive()); + + match (last_day, now_day) { + (Some(last_day), Some(now_day)) => last_day != now_day, + _ => now / 86400 != last / 86400, + } } Rotation::Never => false, } @@ -73,6 +78,7 @@ pub struct RollingAppender { size: u64, // Store as seconds since Unix epoch last_roll_ts: i64, + rotation_retry_cooldown_until: Option, } impl RollingAppender { @@ -119,6 +125,7 @@ impl RollingAppender { file: None, size: 0, last_roll_ts: Zoned::now().timestamp().as_second(), + rotation_retry_cooldown_until: None, }; // Eagerly open the file to validate the path and capture accurate // initial size / last-roll timestamp. @@ -177,6 +184,13 @@ impl RollingAppender { } fn should_roll(&self, write_len: u64) -> bool { + if self + .rotation_retry_cooldown_until + .is_some_and(|deadline| deadline > Instant::now()) + { + return false; + } + // 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 { @@ -256,6 +270,7 @@ impl RollingAppender { // This overrides whatever open_file() derived from mtime, ensuring // we stick to the logical rotation time. self.last_roll_ts = now.timestamp().as_second(); + self.rotation_retry_cooldown_until = None; counter!(METRIC_LOG_CLEANER_ROTATION_TOTAL).increment(1); histogram!(METRIC_LOG_CLEANER_ROTATION_DURATION_SECONDS).record(rotate_started.elapsed().as_secs_f64()); gauge!(METRIC_LOG_CLEANER_ACTIVE_FILE_SIZE_BYTES).set(self.size as f64); @@ -290,6 +305,7 @@ impl RollingAppender { "RollingAppender: Failed to rotate log file after {} retries. Error: {:?}", MAX_RETRIES, last_error ); + self.rotation_retry_cooldown_until = Some(Instant::now() + ROTATION_RETRY_COOLDOWN); counter!(METRIC_LOG_CLEANER_ROTATION_FAILURES_TOTAL).increment(1); histogram!(METRIC_LOG_CLEANER_ROTATION_DURATION_SECONDS).record(rotate_started.elapsed().as_secs_f64()); @@ -541,4 +557,32 @@ mod tests { "size should reflect existing file content" ); } + + #[test] + fn test_daily_rotation_uses_calendar_day_boundaries() { + let now = Zoned::now(); + let same_day_last = now.timestamp().as_second().saturating_sub(60); + assert!(!Rotation::Daily.check_should_roll(same_day_last, now.timestamp().as_second())); + + let previous_day_last = now.timestamp().as_second().saturating_sub(86_400); + let last_day = Local.timestamp_opt(previous_day_last, 0).single().map(|ts| ts.date_naive()); + let now_day = Local + .timestamp_opt(now.timestamp().as_second(), 0) + .single() + .map(|ts| ts.date_naive()); + if last_day != now_day { + assert!(Rotation::Daily.check_should_roll(previous_day_last, now.timestamp().as_second())); + } + } + + #[test] + fn test_rotation_cooldown_skips_immediate_retry() { + let tmp = TempDir::new().unwrap(); + let mut appender = + RollingAppender::new(tmp.path(), "app.log".to_string(), Rotation::Never, 3, FileMatchMode::Suffix).unwrap(); + appender.size = 4; + appender.rotation_retry_cooldown_until = Some(Instant::now() + Duration::from_secs(1)); + + assert!(!appender.should_roll(1)); + } } diff --git a/rustfs/src/profiling.rs b/rustfs/src/profiling.rs index 9bd0e66d5..e721650c1 100644 --- a/rustfs/src/profiling.rs +++ b/rustfs/src/profiling.rs @@ -125,7 +125,17 @@ fn legacy_profiling_env_keys_present() -> Vec<&'static str> { } fn target_env() -> &'static str { - option_env!("CARGO_CFG_TARGET_ENV").unwrap_or("unknown") + if cfg!(target_env = "gnu") { + "gnu" + } else if cfg!(target_env = "musl") { + "musl" + } else if cfg!(target_env = "msvc") { + "msvc" + } else if cfg!(target_env = "sgx") { + "sgx" + } else { + "unknown" + } } #[cfg(test)] @@ -146,4 +156,10 @@ mod tests { }, ); } + + #[test] + fn target_env_reports_known_compile_target_or_unknown() { + let value = target_env(); + assert!(matches!(value, "gnu" | "musl" | "msvc" | "sgx" | "unknown")); + } }