mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-01 09:48:20 +00:00
fix(obs): tighten low-risk telemetry correctness (#4483)
Refs rustfs/backlog#1007 Refs rustfs/backlog#986 - respect explicit log-target directives when injecting noisy-crate suppressions - fix daily rotation day-boundary checks and add a short retry cooldown after failed rotations - preserve recorder metadata across label variants and serialize gauge updates before export - correct profiling target_env reporting, trace all=true parsing, cleaner accounting/docs, and legacy system interval rounding Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
@@ -93,13 +93,6 @@ impl ServiceTraceOpts {
|
|||||||
self.batch_replication = query_pairs.get("batch-replication").is_some_and(|v| v == "true");
|
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_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");
|
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.rebalance = query_pairs.get("rebalance").is_some_and(|v| v == "true");
|
||||||
self.storage = query_pairs.get("storage").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");
|
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.ftp = query_pairs.get("ftp").is_some_and(|v| v == "true");
|
||||||
self.ilm = query_pairs.get("ilm").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") {
|
if let Some(threshold) = query_pairs.get("threshold") {
|
||||||
let duration = parse_duration(threshold)?;
|
let duration = parse_duration(threshold)?;
|
||||||
self.threshold = duration;
|
self.threshold = duration;
|
||||||
@@ -117,3 +117,22 @@ impl ServiceTraceOpts {
|
|||||||
Ok(())
|
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));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -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:
|
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`;
|
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`;
|
3. remove any file whose individual size exceeds `max_single_file_size_bytes`;
|
||||||
4. if compression is enabled, archive before deletion;
|
4. if compression is enabled, archive before deletion;
|
||||||
|
|||||||
@@ -69,7 +69,7 @@ pub struct LogCleaner {
|
|||||||
pub(super) active_filename: String,
|
pub(super) active_filename: String,
|
||||||
/// Whether `file_pattern` is interpreted as a prefix or suffix.
|
/// Whether `file_pattern` is interpreted as a prefix or suffix.
|
||||||
pub(super) match_mode: FileMatchMode,
|
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,
|
pub(super) keep_files: usize,
|
||||||
/// Optional cap for the cumulative size of regular logs.
|
/// Optional cap for the cumulative size of regular logs.
|
||||||
pub(super) max_total_size_bytes: u64,
|
pub(super) max_total_size_bytes: u64,
|
||||||
@@ -205,8 +205,8 @@ impl LogCleaner {
|
|||||||
/// Choose regular log files that should be compressed and/or deleted.
|
/// Choose regular log files that should be compressed and/or deleted.
|
||||||
///
|
///
|
||||||
/// The `files` slice must already be sorted from oldest to newest. The
|
/// The `files` slice must already be sorted from oldest to newest. The
|
||||||
/// method first preserves the newest `keep_files` generations, then applies
|
/// method first enforces the `keep_files` ceiling, then applies total-size
|
||||||
/// total-size and per-file-size limits to the remaining tail.
|
/// and per-file-size limits to the remaining tail.
|
||||||
pub(super) fn select_files_to_process(&self, files: &[FileInfo], total_size: u64) -> Vec<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_delete = Vec::new();
|
||||||
if files.is_empty() {
|
if files.is_empty() {
|
||||||
@@ -325,8 +325,6 @@ impl LogCleaner {
|
|||||||
} else {
|
} else {
|
||||||
match injector.steal_batch_and_pop(&local_worker) {
|
match injector.steal_batch_and_pop(&local_worker) {
|
||||||
Steal::Success(file) => {
|
Steal::Success(file) => {
|
||||||
attempts.fetch_add(1, Ordering::Relaxed);
|
|
||||||
successes.fetch_add(1, Ordering::Relaxed);
|
|
||||||
Some(file)
|
Some(file)
|
||||||
}
|
}
|
||||||
Steal::Retry => continue,
|
Steal::Retry => continue,
|
||||||
@@ -357,6 +355,7 @@ impl LogCleaner {
|
|||||||
continue;
|
continue;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
let mut file = file;
|
||||||
let compressed = match compress_file(&file.path, &options) {
|
let compressed = match compress_file(&file.path, &options) {
|
||||||
Ok(output) => {
|
Ok(output) => {
|
||||||
debug!(
|
debug!(
|
||||||
@@ -371,6 +370,7 @@ impl LogCleaner {
|
|||||||
state = "parallel_compression_done",
|
state = "parallel_compression_done",
|
||||||
"log cleaner state changed"
|
"log cleaner state changed"
|
||||||
);
|
);
|
||||||
|
file.projected_freed_bytes = output.input_bytes.saturating_sub(output.output_bytes);
|
||||||
true
|
true
|
||||||
}
|
}
|
||||||
Err(err) => {
|
Err(err) => {
|
||||||
@@ -508,7 +508,9 @@ impl LogCleaner {
|
|||||||
state = "serial_compression_done",
|
state = "serial_compression_done",
|
||||||
"log cleaner state changed"
|
"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) => {
|
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");
|
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 {
|
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");
|
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;
|
deleted += 1;
|
||||||
freed += f.size;
|
freed += f.projected_freed_bytes;
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
match self.secure_delete(&f.path) {
|
match self.secure_delete(&f.path) {
|
||||||
Ok(()) => {
|
Ok(()) => {
|
||||||
deleted += 1;
|
deleted += 1;
|
||||||
freed += f.size;
|
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, "log cleaner state changed");
|
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) => {
|
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");
|
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
|
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 {
|
pub fn keep_files(mut self, keep_files: usize) -> Self {
|
||||||
self.keep_files = keep_files;
|
self.keep_files = keep_files;
|
||||||
self
|
self
|
||||||
|
|||||||
@@ -127,8 +127,6 @@ mod tests {
|
|||||||
fn assert_parallel_cleanup_completes(file_count: usize, workers: usize) -> std::io::Result<()> {
|
fn assert_parallel_cleanup_completes(file_count: usize, workers: usize) -> std::io::Result<()> {
|
||||||
let tmp = TempDir::new()?;
|
let tmp = TempDir::new()?;
|
||||||
let dir = tmp.path().to_path_buf();
|
let dir = tmp.path().to_path_buf();
|
||||||
let expected_freed = (file_count * 256) as u64;
|
|
||||||
|
|
||||||
for i in 0..file_count {
|
for i in 0..file_count {
|
||||||
create_log_file(&dir, &format!("app.log.2024-01-{i:03}"), 256)?;
|
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))
|
name.starts_with("app.log.") && !compressed_suffixes.iter().any(|suffix| name.ends_with(suffix))
|
||||||
})
|
})
|
||||||
.count();
|
.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.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!(archive_count, file_count, "each rotated log should leave behind one archive");
|
||||||
assert_eq!(original_count, 0, "compressed source logs should be removed");
|
assert_eq!(original_count, 0, "compressed source logs should be removed");
|
||||||
Ok(())
|
Ok(())
|
||||||
|
|||||||
@@ -199,6 +199,7 @@ pub(super) fn scan_log_directory(
|
|||||||
let info = FileInfo {
|
let info = FileInfo {
|
||||||
path,
|
path,
|
||||||
size: file_size,
|
size: file_size,
|
||||||
|
projected_freed_bytes: file_size,
|
||||||
modified,
|
modified,
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|||||||
@@ -176,6 +176,11 @@ pub(super) struct FileInfo {
|
|||||||
///
|
///
|
||||||
/// This value is used for retention accounting and freed-byte metrics.
|
/// This value is used for retention accounting and freed-byte metrics.
|
||||||
pub size: u64,
|
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.
|
/// Last-modification timestamp from the filesystem.
|
||||||
///
|
///
|
||||||
/// The selection phase sorts on this timestamp so the oldest files are
|
/// The selection phase sorts on this timestamp so the oldest files are
|
||||||
|
|||||||
@@ -80,7 +80,7 @@ const LEGACY_ENV_OBS_PROFILING_ENABLED: &str = "RUSTFS_OBS_PROFILING_ENABLED";
|
|||||||
/// - All fields are `Option<T>` to allow partial configuration via environment
|
/// - All fields are `Option<T>` to allow partial configuration via environment
|
||||||
/// variables with sensible defaults provided by constants in `rustfs-config`.
|
/// variables with sensible defaults provided by constants in `rustfs-config`.
|
||||||
/// - `log_keep_files` is used to derive the rolling-appender's upper bound on
|
/// - `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
|
/// # Example
|
||||||
/// ```no_run
|
/// ```no_run
|
||||||
|
|||||||
@@ -403,7 +403,7 @@ fn parse_metrics_interval(primary_env: &str, legacy_env: &str, default_interval:
|
|||||||
|
|
||||||
fn parse_system_metrics_interval() -> Duration {
|
fn parse_system_metrics_interval() -> Duration {
|
||||||
get_env_opt_u64(ENV_SYSTEM_METRICS_INTERVAL)
|
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))
|
.or_else(|| get_env_opt_u64(ENV_DEFAULT_METRICS_INTERVAL))
|
||||||
.filter(|&v| v > 0)
|
.filter(|&v| v > 0)
|
||||||
.map(Duration::from_secs)
|
.map(Duration::from_secs)
|
||||||
@@ -1280,6 +1280,10 @@ pub fn init_metrics_runtime(token: CancellationToken) {
|
|||||||
let mut next_resource_run = now;
|
let mut next_resource_run = now;
|
||||||
let mut next_system_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")]
|
#[cfg(feature = "gpu")]
|
||||||
let current_pid = match sysinfo::get_current_pid() {
|
let current_pid = match sysinfo::get_current_pid() {
|
||||||
Ok(pid) => Some(pid),
|
Ok(pid) => Some(pid),
|
||||||
@@ -1800,4 +1804,32 @@ mod tests {
|
|||||||
expire_series_zero_tombstones(&mut zero_tombstones);
|
expire_series_zero_tombstones(&mut zero_tombstones);
|
||||||
assert!(zero_tombstones.is_empty());
|
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));
|
||||||
|
},
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -150,6 +150,16 @@ fn should_demote_http_request_logs(logger_level: &str, default_level: Option<&st
|
|||||||
matches!(level.as_str(), "info" | "warn")
|
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 {
|
pub(super) fn build_env_filter(logger_level: &str, default_level: Option<&str>) -> EnvFilter {
|
||||||
// 1. Determine the base filter source.
|
// 1. Determine the base filter source.
|
||||||
// If `default_level` is set (e.g. forced override), we use it.
|
// 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 {
|
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.
|
// We use `add_directive` which effectively appends to the filter.
|
||||||
// If RUST_LOG already specified `hyper=debug`, adding `hyper=off` later MIGHT override it
|
// 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.
|
// 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}"
|
||||||
|
);
|
||||||
|
});
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -111,7 +111,7 @@ impl Builder {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug)]
|
#[derive(Debug, Clone)]
|
||||||
struct MetricMetadata {
|
struct MetricMetadata {
|
||||||
unit: Option<Unit>,
|
unit: Option<Unit>,
|
||||||
description: SharedString,
|
description: SharedString,
|
||||||
@@ -196,7 +196,7 @@ impl Recorder {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn get_metadata_for_builder(&self, key_name: &str) -> Option<MetricMetadata> {
|
fn get_metadata_for_builder(&self, key_name: &str) -> Option<MetricMetadata> {
|
||||||
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,
|
gauge,
|
||||||
labels,
|
labels,
|
||||||
value: AtomicU64::new(0),
|
value: AtomicU64::new(0),
|
||||||
|
record_lock: Mutex::new(()),
|
||||||
}));
|
}));
|
||||||
|
|
||||||
Self::insert_cached_metric(&self.cached_gauges, key.clone(), handle, "gauge")
|
Self::insert_cached_metric(&self.cached_gauges, key.clone(), handle, "gauge")
|
||||||
@@ -313,10 +314,12 @@ struct WrappedGauge {
|
|||||||
gauge: opentelemetry::metrics::Gauge<f64>,
|
gauge: opentelemetry::metrics::Gauge<f64>,
|
||||||
labels: Vec<KeyValue>,
|
labels: Vec<KeyValue>,
|
||||||
value: AtomicU64,
|
value: AtomicU64,
|
||||||
|
record_lock: Mutex<()>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl GaugeFn for WrappedGauge {
|
impl GaugeFn for WrappedGauge {
|
||||||
fn increment(&self, value: f64) {
|
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 current = self.value.load(Ordering::Relaxed);
|
||||||
let mut new = f64::from_bits(current) + value;
|
let mut new = f64::from_bits(current) + value;
|
||||||
while let Err(val) = self
|
while let Err(val) = self
|
||||||
@@ -331,6 +334,7 @@ impl GaugeFn for WrappedGauge {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn decrement(&self, value: f64) {
|
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 current = self.value.load(Ordering::Relaxed);
|
||||||
let mut new = f64::from_bits(current) - value;
|
let mut new = f64::from_bits(current) - value;
|
||||||
while let Err(val) = self
|
while let Err(val) = self
|
||||||
@@ -345,6 +349,7 @@ impl GaugeFn for WrappedGauge {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn set(&self, value: f64) {
|
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.value.store(value.to_bits(), Ordering::Relaxed);
|
||||||
self.gauge.record(value, &self.labels);
|
self.gauge.record(value, &self.labels);
|
||||||
}
|
}
|
||||||
@@ -473,4 +478,25 @@ mod tests {
|
|||||||
let cache = shared.cached_counters.read().unwrap();
|
let cache = shared.cached_counters.read().unwrap();
|
||||||
assert_eq!(cache.len(), 1, "concurrent registrations should produce exactly one cache entry");
|
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);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -23,6 +23,7 @@ use crate::global::{
|
|||||||
METRIC_LOG_CLEANER_ACTIVE_FILE_SIZE_BYTES, METRIC_LOG_CLEANER_ROTATION_DURATION_SECONDS,
|
METRIC_LOG_CLEANER_ACTIVE_FILE_SIZE_BYTES, METRIC_LOG_CLEANER_ROTATION_DURATION_SECONDS,
|
||||||
METRIC_LOG_CLEANER_ROTATION_FAILURES_TOTAL, METRIC_LOG_CLEANER_ROTATION_TOTAL,
|
METRIC_LOG_CLEANER_ROTATION_FAILURES_TOTAL, METRIC_LOG_CLEANER_ROTATION_TOTAL,
|
||||||
};
|
};
|
||||||
|
use chrono::{Local, TimeZone as _};
|
||||||
use jiff::Zoned;
|
use jiff::Zoned;
|
||||||
use metrics::{counter, gauge, histogram};
|
use metrics::{counter, gauge, histogram};
|
||||||
use std::fs::{self, File};
|
use std::fs::{self, File};
|
||||||
@@ -30,7 +31,7 @@ use std::io::{self, Write};
|
|||||||
use std::path::{Path, PathBuf};
|
use std::path::{Path, PathBuf};
|
||||||
use std::sync::atomic::{AtomicU64, Ordering};
|
use std::sync::atomic::{AtomicU64, Ordering};
|
||||||
use std::thread;
|
use std::thread;
|
||||||
use std::time::Duration;
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
#[derive(Debug, Clone, Copy)]
|
#[derive(Debug, Clone, Copy)]
|
||||||
pub enum Rotation {
|
pub enum Rotation {
|
||||||
@@ -45,6 +46,7 @@ pub enum Rotation {
|
|||||||
/// may otherwise collide when multiple rotations occur within the same
|
/// may otherwise collide when multiple rotations occur within the same
|
||||||
/// timestamp tick.
|
/// timestamp tick.
|
||||||
static ROLL_UNIQUIFIER: AtomicU64 = AtomicU64::new(0);
|
static ROLL_UNIQUIFIER: AtomicU64 = AtomicU64::new(0);
|
||||||
|
const ROTATION_RETRY_COOLDOWN: Duration = Duration::from_secs(5);
|
||||||
|
|
||||||
impl Rotation {
|
impl Rotation {
|
||||||
fn check_should_roll(&self, last: i64, now: i64) -> bool {
|
fn check_should_roll(&self, last: i64, now: i64) -> bool {
|
||||||
@@ -52,10 +54,13 @@ impl Rotation {
|
|||||||
Rotation::Minutely => now / 60 != last / 60,
|
Rotation::Minutely => now / 60 != last / 60,
|
||||||
Rotation::Hourly => now / 3600 != last / 3600,
|
Rotation::Hourly => now / 3600 != last / 3600,
|
||||||
Rotation::Daily => {
|
Rotation::Daily => {
|
||||||
// Align daily rotation with the local day boundary rather than UTC midnight.
|
let last_day = Local.timestamp_opt(last, 0).single().map(|ts| ts.date_naive());
|
||||||
// We shift both timestamps by the current local offset before bucketing into days.
|
let now_day = Local.timestamp_opt(now, 0).single().map(|ts| ts.date_naive());
|
||||||
let offset_secs = Zoned::now().offset().seconds() as i64;
|
|
||||||
(now + offset_secs) / 86400 != (last + offset_secs) / 86400
|
match (last_day, now_day) {
|
||||||
|
(Some(last_day), Some(now_day)) => last_day != now_day,
|
||||||
|
_ => now / 86400 != last / 86400,
|
||||||
|
}
|
||||||
}
|
}
|
||||||
Rotation::Never => false,
|
Rotation::Never => false,
|
||||||
}
|
}
|
||||||
@@ -73,6 +78,7 @@ pub struct RollingAppender {
|
|||||||
size: u64,
|
size: u64,
|
||||||
// Store as seconds since Unix epoch
|
// Store as seconds since Unix epoch
|
||||||
last_roll_ts: i64,
|
last_roll_ts: i64,
|
||||||
|
rotation_retry_cooldown_until: Option<Instant>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl RollingAppender {
|
impl RollingAppender {
|
||||||
@@ -119,6 +125,7 @@ impl RollingAppender {
|
|||||||
file: None,
|
file: None,
|
||||||
size: 0,
|
size: 0,
|
||||||
last_roll_ts: Zoned::now().timestamp().as_second(),
|
last_roll_ts: Zoned::now().timestamp().as_second(),
|
||||||
|
rotation_retry_cooldown_until: None,
|
||||||
};
|
};
|
||||||
// Eagerly open the file to validate the path and capture accurate
|
// Eagerly open the file to validate the path and capture accurate
|
||||||
// initial size / last-roll timestamp.
|
// initial size / last-roll timestamp.
|
||||||
@@ -177,6 +184,13 @@ impl RollingAppender {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn should_roll(&self, write_len: u64) -> bool {
|
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)
|
// 1. Size-based check (Cheap, check first)
|
||||||
// If max_size is set (non-zero) and writing would exceed it, roll immediately.
|
// 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 {
|
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
|
// This overrides whatever open_file() derived from mtime, ensuring
|
||||||
// we stick to the logical rotation time.
|
// we stick to the logical rotation time.
|
||||||
self.last_roll_ts = now.timestamp().as_second();
|
self.last_roll_ts = now.timestamp().as_second();
|
||||||
|
self.rotation_retry_cooldown_until = None;
|
||||||
counter!(METRIC_LOG_CLEANER_ROTATION_TOTAL).increment(1);
|
counter!(METRIC_LOG_CLEANER_ROTATION_TOTAL).increment(1);
|
||||||
histogram!(METRIC_LOG_CLEANER_ROTATION_DURATION_SECONDS).record(rotate_started.elapsed().as_secs_f64());
|
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);
|
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: {:?}",
|
"RollingAppender: Failed to rotate log file after {} retries. Error: {:?}",
|
||||||
MAX_RETRIES, last_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);
|
counter!(METRIC_LOG_CLEANER_ROTATION_FAILURES_TOTAL).increment(1);
|
||||||
histogram!(METRIC_LOG_CLEANER_ROTATION_DURATION_SECONDS).record(rotate_started.elapsed().as_secs_f64());
|
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"
|
"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));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+17
-1
@@ -125,7 +125,17 @@ fn legacy_profiling_env_keys_present() -> Vec<&'static str> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn target_env() -> &'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)]
|
#[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"));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user