diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 2e0c38e9a..df6cfad62 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -284,9 +284,9 @@ anyhow::Result // in library code (OK in tests/CLI) ### Metrics -- Prometheus-style metrics via `rustfs-metrics` crate +- Prometheus-style metrics via `rustfs-obs` runtime and schema - I/O-specific counters via `rustfs-io-metrics` -- Registration happens at crate level, collection in `metrics` crate +- Registration happens at crate level, collection/reporting in `rustfs-obs` ### Testing @@ -368,7 +368,7 @@ The binary (`main.rs`) boots in this order: Add handler in `admin/handlers/`, register in `admin/router.rs` - **"Where do I add a new metric?"** - Define in `crates/metrics/`, register collector, expose via `/minio/v2/metrics` + Define descriptor/collector in `crates/obs/src/metrics/`, expose via `/minio/v2/metrics` --- diff --git a/Cargo.lock b/Cargo.lock index 587211b45..a630b93e4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -830,9 +830,9 @@ dependencies = [ [[package]] name = "aws-sdk-sts" -version = "1.101.0" +version = "1.102.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ab41ad64e4051ecabeea802d6a17845a91e83287e1dd249e6963ea1ba78c428a" +checksum = "0fc35b7a14cabdad13795fbbbd26d5ddec0882c01492ceedf2af575aad5f37dd" dependencies = [ "aws-credential-types", "aws-runtime", @@ -3067,20 +3067,33 @@ dependencies = [ ] [[package]] -name = "dial9-tokio-telemetry" -version = "0.2.1" +name = "dial9-macro" +version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "35c8348045a09a7dee69d000124d42f773f59f85737d3212ae65aa64ac0b4e3e" +checksum = "877b67d2a742d9c722961c468e8b9cfe87d78e53f2e1d61125c8d388bb8a3bba" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + +[[package]] +name = "dial9-tokio-telemetry" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1edba70ebc611e0ebc889d7f6b3d08ce742e48cb188da1f298ec6378b0b4f86b" dependencies = [ "arc-swap", "bon", "crossbeam-queue", + "dial9-macro", "dial9-trace-format", "flate2", "futures-util", "hostname", "libc", "metrique", + "metrique-timesource", "metrique-writer", "pin-project-lite", "serde", @@ -3093,9 +3106,9 @@ dependencies = [ [[package]] name = "dial9-trace-format" -version = "0.2.0" +version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "80e0ee560b05f09bf817602d57644947e31e83c521d4e0277f723a6e64d44f92" +checksum = "8c9d20951a4584f9ea544ff27143709821c4709f27d30b52395300f020392475" dependencies = [ "dial9-trace-format-derive", "serde", @@ -3103,9 +3116,9 @@ dependencies = [ [[package]] name = "dial9-trace-format-derive" -version = "0.2.0" +version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9dbbd8126d4d6613931317cfe2a7275c1cd487e41c961e42456ab5f956570030" +checksum = "5800a5a155746fe428681cdc30570f33d961a31ed800e6df41b78d0af64529ed" dependencies = [ "proc-macro2", "quote", @@ -5018,9 +5031,9 @@ dependencies = [ [[package]] name = "libbz2-rs-sys" -version = "0.2.2" +version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2c4a545a15244c7d945065b5d392b2d2d7f21526fba56ce51467b06ed445e8f7" +checksum = "b3a6a8c165077efc8f3a971534c50ea6a1a18b329ef4a66e897a7e3a1494565f" [[package]] name = "libc" @@ -7722,7 +7735,6 @@ dependencies = [ "rustfs-kms", "rustfs-lock", "rustfs-madmin", - "rustfs-metrics", "rustfs-notify", "rustfs-object-capacity", "rustfs-obs", @@ -8085,6 +8097,7 @@ dependencies = [ "criterion", "metrics", "num_cpus", + "sysinfo", "thiserror 2.0.18", "tokio", "tracing", @@ -8152,6 +8165,7 @@ dependencies = [ "crossbeam-queue", "futures", "parking_lot 0.12.5", + "rustfs-io-metrics", "rustfs-utils", "serde", "serde_json", @@ -8176,22 +8190,6 @@ dependencies = [ "time", ] -[[package]] -name = "rustfs-metrics" -version = "0.0.5" -dependencies = [ - "metrics", - "nvml-wrapper", - "rustfs-config", - "rustfs-ecstore", - "rustfs-utils", - "sysinfo", - "thiserror 2.0.18", - "tokio", - "tokio-util", - "tracing", -] - [[package]] name = "rustfs-notify" version = "0.0.5" @@ -8256,6 +8254,7 @@ dependencies = [ "jiff", "metrics", "num_cpus", + "nvml-wrapper", "opentelemetry", "opentelemetry-appender-tracing", "opentelemetry-otlp", @@ -8263,13 +8262,19 @@ dependencies = [ "opentelemetry-stdout", "opentelemetry_sdk", "pyroscope", + "rustfs-audit", "rustfs-config", + "rustfs-ecstore", + "rustfs-io-metrics", + "rustfs-notify", "rustfs-utils", "serde", + "sysinfo", "temp-env", "tempfile", "thiserror 2.0.18", "tokio", + "tokio-util", "tracing", "tracing-appender", "tracing-error", @@ -9570,9 +9575,9 @@ dependencies = [ [[package]] name = "symbolic-common" -version = "12.18.0" +version = "12.18.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6aba7211a1803a826f108af9f4d86d25abe880712f2a6449479279e861b293f8" +checksum = "b1f3cdeaae6779ecba2567f20bf7716718b8c4ce6717c9def4ced18786bb11ea" dependencies = [ "debugid", "memmap2 0.9.10", @@ -9582,15 +9587,21 @@ dependencies = [ [[package]] name = "symbolic-demangle" -version = "12.18.0" +version = "12.18.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "595bddd9d363c2ef6fc9fb33b98416ff209c5aae8bdca89a7dbbf2ef5e0ecc45" +checksum = "672c6ad9cb8fce6a1283cc9df9070073cccad00ae241b80e3686328a64e3523b" dependencies = [ "cpp_demangle", "rustc-demangle", "symbolic-common", ] +[[package]] +name = "symlink" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7973cce6668464ea31f176d85b13c7ab3bba2cb3b77a2ed26abd7801688010a" + [[package]] name = "syn" version = "1.0.109" @@ -10130,11 +10141,12 @@ dependencies = [ [[package]] name = "tracing-appender" -version = "0.2.4" +version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "786d480bce6247ab75f005b14ae1624ad978d3029d9113f0a22fa1ac773faeaf" +checksum = "050686193eb999b4bb3bc2acfa891a13da00f79734704c4b8b4ef1a10b368a3c" dependencies = [ "crossbeam-channel", + "symlink", "thiserror 2.0.18", "time", "tracing-subscriber", @@ -10483,11 +10495,11 @@ checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" [[package]] name = "wasip2" -version = "1.0.2+wasi-0.2.9" +version = "1.0.3+wasi-0.2.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9517f9239f02c069db75e65f174b3da828fe5f5b945c4dd26bd25d89c03ebcf5" +checksum = "20064672db26d7cdc89c7798c48a0fdfac8213434a1186e5ef29fd560ae223d6" dependencies = [ - "wit-bindgen", + "wit-bindgen 0.57.1", ] [[package]] @@ -10496,7 +10508,7 @@ version = "0.4.0+wasi-0.3.0-rc-2026-01-06" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5428f8bf88ea5ddc08faddef2ac4a67e390b88186c703ce6dbd955e1c145aca5" dependencies = [ - "wit-bindgen", + "wit-bindgen 0.51.0", ] [[package]] @@ -11052,6 +11064,12 @@ dependencies = [ "wit-bindgen-rust-macro", ] +[[package]] +name = "wit-bindgen" +version = "0.57.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" + [[package]] name = "wit-bindgen-core" version = "0.51.0" diff --git a/Cargo.toml b/Cargo.toml index 36c414f56..46907da2a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,7 +31,6 @@ members = [ "crates/kms", # Key Management Service "crates/lock", # Distributed locking implementation "crates/madmin", # Management dashboard and admin API interface - "crates/metrics", # Metrics collection and reporting "crates/notify", # Notification system for events "crates/obs", # Observability utilities "crates/object-capacity", # Capacity scan and refresh core @@ -94,7 +93,6 @@ rustfs-keystone = { path = "crates/keystone", version = "0.0.5" } rustfs-kms = { path = "crates/kms", version = "0.0.5" } rustfs-lock = { path = "crates/lock", version = "0.0.5" } rustfs-madmin = { path = "crates/madmin", version = "0.0.5" } -rustfs-metrics = { path = "crates/metrics", version = "0.0.5" } rustfs-notify = { path = "crates/notify", version = "0.0.5" } rustfs-io-metrics = { path = "crates/io-metrics", version = "0.0.5" } rustfs-io-core = { path = "crates/io-core", version = "0.0.5" } @@ -266,7 +264,7 @@ tempfile = "3.27.0" test-case = "3.3.1" thiserror = "2.0.18" tracing = { version = "0.1.44" } -tracing-appender = "0.2.4" +tracing-appender = "0.2.5" tracing-error = "0.2.1" tracing-opentelemetry = "0.32.1" tracing-subscriber = { version = "0.3.23", features = ["env-filter", "time"] } @@ -284,7 +282,7 @@ zstd = "0.13.3" # Observability and Metrics metrics = "0.24.3" -dial9-tokio-telemetry = "0.2" +dial9-tokio-telemetry = "0.3" opentelemetry = { version = "0.31.0" } opentelemetry-appender-tracing = { version = "0.31.1", features = ["experimental_use_tracing_span_context", "experimental_metadata_attributes", "spec_unstable_logs_enabled"] } opentelemetry-otlp = { version = "0.31.1", features = ["gzip-http", "reqwest-rustls"] } diff --git a/crates/audit/src/global.rs b/crates/audit/src/global.rs index 4031a3847..2519e1174 100644 --- a/crates/audit/src/global.rs +++ b/crates/audit/src/global.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::{AuditEntry, AuditResult, AuditSystem}; +use crate::{AuditEntry, AuditResult, AuditSystem, system::AuditTargetMetricSnapshot}; use rustfs_ecstore::config::Config; use std::sync::{Arc, OnceLock}; use tracing::{debug, error, trace, warn}; @@ -89,6 +89,15 @@ pub async fn reload_audit_config(config: Config) -> AuditResult<()> { with_audit_system!(|system: Arc| async move { system.reload_config(config).await }) } +/// Returns per-target audit delivery metrics for Prometheus collection. +pub async fn audit_target_metrics() -> Vec { + if let Some(system) = audit_system() { + system.snapshot_target_metrics().await + } else { + Vec::new() + } +} + /// Check if the global audit system is running pub async fn is_audit_system_running() -> bool { if let Some(system) = audit_system() { diff --git a/crates/audit/src/lib.rs b/crates/audit/src/lib.rs index 7cca0063b..2efd99cd7 100644 --- a/crates/audit/src/lib.rs +++ b/crates/audit/src/lib.rs @@ -31,4 +31,4 @@ pub use error::{AuditError, AuditResult}; pub use global::*; pub use observability::{AuditMetrics, AuditMetricsReport, PerformanceValidation}; pub use registry::AuditRegistry; -pub use system::AuditSystem; +pub use system::{AuditSystem, AuditTargetMetricSnapshot}; diff --git a/crates/audit/src/system.rs b/crates/audit/src/system.rs index d47729722..2c050a682 100644 --- a/crates/audit/src/system.rs +++ b/crates/audit/src/system.rs @@ -24,6 +24,14 @@ use std::sync::Arc; use tokio::sync::{Mutex, RwLock, mpsc}; use tracing::{error, info, warn}; +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct AuditTargetMetricSnapshot { + pub failed_messages: u64, + pub queue_length: u64, + pub target_id: String, + pub total_messages: u64, +} + /// State of the audit system #[derive(Debug, Clone, PartialEq, Eq)] pub enum AuditSystemState { @@ -537,8 +545,14 @@ impl AuditSystem { TargetError::Timeout(_) => { warn!("Timeout sending to target {}, retrying...", target.id()); } + TargetError::Dropped(reason) => { + warn!("Dropped queued payload for target {}: {}", target.id(), reason); + observability::record_target_failure(); + break; + } _ => { error!("Permanent error for target {}: {}", target.id(), e); + target.record_final_failure(); observability::record_target_failure(); break; } @@ -552,6 +566,7 @@ impl AuditSystem { if retries >= MAX_RETRIES && !success { warn!("Max retries exceeded for key {}, target: {}, skipping", key.to_string(), target.id()); + target.record_final_failure(); observability::record_target_failure(); } } @@ -664,6 +679,25 @@ impl AuditSystem { registry.list_target_values() } + /// Returns per-target delivery metrics for Prometheus collection. + pub async fn snapshot_target_metrics(&self) -> Vec { + let targets = self.get_target_values().await; + let mut snapshots = Vec::with_capacity(targets.len()); + + for target in targets { + let delivery = target.delivery_snapshot(); + snapshots.push(AuditTargetMetricSnapshot { + failed_messages: delivery.failed_messages, + queue_length: delivery.queue_length, + target_id: target.id().to_string(), + total_messages: delivery.total_messages, + }); + } + + snapshots.sort_by(|a, b| a.target_id.cmp(&b.target_id)); + snapshots + } + /// Gets information about a specific target /// /// # Arguments diff --git a/crates/io-metrics/Cargo.toml b/crates/io-metrics/Cargo.toml index cc920732e..08e957dfa 100644 --- a/crates/io-metrics/Cargo.toml +++ b/crates/io-metrics/Cargo.toml @@ -34,6 +34,7 @@ num_cpus = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true, features = ["sync","rt"] } tracing = { workspace = true } +sysinfo = { workspace = true } [dev-dependencies] criterion = { workspace = true } diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index 4a175feea..1c2c5a636 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -61,6 +61,8 @@ pub mod deadlock_metrics; pub mod io_metrics; pub mod lock_metrics; pub mod performance; +pub mod process_lock_metrics; +pub mod sampler; pub mod timeout_metrics; pub use autotuner::{AutoTuner, TunerConfig, TuningResult}; @@ -108,6 +110,16 @@ pub use lock_metrics::{ record_spin_attempt, record_spin_count_change, }; +pub use process_lock_metrics::{ + ProcessLockSnapshot, ProcessPlatformSnapshot, record_read_lock_held_acquire, record_read_lock_held_release, + record_write_lock_held_acquire, record_write_lock_held_release, snapshot_process_lock_counts, + snapshot_process_platform_stats, +}; +pub use sampler::{ + ProcessResourceSnapshot, ProcessStatusSnapshot, ProcessSystemSnapshot, snapshot_process_platform, snapshot_process_resource, + snapshot_process_resource_and_system, snapshot_process_system, +}; + // Timeout metrics exports pub use timeout_metrics::{ TimeoutMetricsSummary, record_dynamic_timeout, record_operation_completion, record_operation_duration, diff --git a/crates/io-metrics/src/process_lock_metrics.rs b/crates/io-metrics/src/process_lock_metrics.rs new file mode 100644 index 000000000..d5b2b146a --- /dev/null +++ b/crates/io-metrics/src/process_lock_metrics.rs @@ -0,0 +1,373 @@ +// 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. + +use std::sync::atomic::{AtomicU64, Ordering}; + +#[cfg(target_os = "linux")] +use std::fs; +#[cfg(any( + target_os = "windows", + target_os = "macos", + target_os = "freebsd", + target_os = "openbsd", + target_os = "netbsd", + target_os = "dragonfly" +))] +use std::process::Command; + +static READ_LOCKS_HELD: AtomicU64 = AtomicU64::new(0); +static WRITE_LOCKS_HELD: AtomicU64 = AtomicU64::new(0); + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct ProcessLockSnapshot { + pub read_locks_held: u64, + pub write_locks_held: u64, +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct ProcessPlatformSnapshot { + pub io_rchar_bytes: Option, + pub io_read_bytes: Option, + pub io_wchar_bytes: Option, + pub io_write_bytes: Option, + pub syscall_read_total: Option, + pub syscall_write_total: Option, + pub virtual_memory_max_bytes: Option, +} + +#[inline(always)] +pub fn record_read_lock_held_acquire() { + READ_LOCKS_HELD.fetch_add(1, Ordering::Relaxed); +} + +#[inline(always)] +pub fn record_write_lock_held_acquire() { + WRITE_LOCKS_HELD.fetch_add(1, Ordering::Relaxed); +} + +#[inline(always)] +pub fn record_read_lock_held_release() { + let _ = READ_LOCKS_HELD.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| Some(value.saturating_sub(1))); +} + +#[inline(always)] +pub fn record_write_lock_held_release() { + let _ = WRITE_LOCKS_HELD.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| Some(value.saturating_sub(1))); +} + +#[inline(always)] +pub fn snapshot_process_lock_counts() -> ProcessLockSnapshot { + ProcessLockSnapshot { + read_locks_held: READ_LOCKS_HELD.load(Ordering::Relaxed), + write_locks_held: WRITE_LOCKS_HELD.load(Ordering::Relaxed), + } +} + +#[inline] +pub fn snapshot_process_platform_stats() -> ProcessPlatformSnapshot { + platform::snapshot() +} + +#[cfg(any( + target_os = "windows", + target_os = "macos", + target_os = "freebsd", + target_os = "openbsd", + target_os = "netbsd", + target_os = "dragonfly" +))] +fn run_command(command: &str, args: &[&str]) -> Option { + let output = Command::new(command).args(args).output().ok()?; + if !output.status.success() { + return None; + } + + let stdout = String::from_utf8_lossy(&output.stdout).trim().to_string(); + if stdout.is_empty() { None } else { Some(stdout) } +} + +#[cfg(any(target_os = "windows", test))] +fn parse_kv_u64(content: &str, key: &str) -> Option { + for line in content.lines() { + let Some((k, v)) = line.split_once(':') else { + continue; + }; + if k.trim().eq_ignore_ascii_case(key) { + return v.trim().parse::().ok(); + } + } + None +} + +#[cfg(target_os = "linux")] +mod platform { + use super::*; + + pub(super) fn snapshot() -> ProcessPlatformSnapshot { + let io = fs::read_to_string("/proc/self/io").ok(); + let status = fs::read_to_string("/proc/self/status").ok(); + let io_stats = io.as_deref().map(parse_proc_self_io).unwrap_or_default(); + + ProcessPlatformSnapshot { + io_rchar_bytes: io_stats.rchar_bytes, + io_read_bytes: io_stats.read_bytes, + io_wchar_bytes: io_stats.wchar_bytes, + io_write_bytes: io_stats.write_bytes, + syscall_read_total: io_stats.syscall_read_total, + syscall_write_total: io_stats.syscall_write_total, + virtual_memory_max_bytes: status.as_deref().and_then(parse_vm_peak_bytes), + } + } + + #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] + struct ProcSelfIoStats { + rchar_bytes: Option, + read_bytes: Option, + syscall_read_total: Option, + syscall_write_total: Option, + wchar_bytes: Option, + write_bytes: Option, + } + + fn parse_proc_self_io(content: &str) -> ProcSelfIoStats { + let mut stats = ProcSelfIoStats::default(); + + for line in content.lines() { + let Some((key, value)) = line.split_once(':') else { + continue; + }; + let Ok(value) = value.trim().parse::() else { + continue; + }; + + match key.trim() { + "rchar" => stats.rchar_bytes = Some(value), + "read_bytes" => stats.read_bytes = Some(value), + "wchar" => stats.wchar_bytes = Some(value), + "write_bytes" => stats.write_bytes = Some(value), + "syscr" => stats.syscall_read_total = Some(value), + "syscw" => stats.syscall_write_total = Some(value), + _ => {} + } + } + + stats + } + + fn parse_vm_peak_bytes(content: &str) -> Option { + for line in content.lines() { + let Some(rest) = line.strip_prefix("VmPeak:") else { + continue; + }; + let mut parts = rest.split_whitespace(); + let value = parts.next()?.parse::().ok()?; + let unit = parts.next().unwrap_or_default(); + return Some(match unit { + "kB" | "KB" | "kb" => value.saturating_mul(1024), + _ => value, + }); + } + None + } + + #[cfg(test)] + mod tests { + use super::*; + + #[test] + fn parse_vm_peak() { + let status = "Name:\trustfs\nVmPeak:\t 2048 kB\nVmRSS:\t 1024 kB\n"; + assert_eq!(parse_vm_peak_bytes(status), Some(2048 * 1024)); + } + + #[test] + fn parse_proc_self_io_extracts_expected_fields() { + let stats = parse_proc_self_io( + "rchar: 11\nwchar: 22\nsyscr: 33\nsyscw: 44\nread_bytes: 55\nwrite_bytes: 66\ncancelled_write_bytes: 77\n", + ); + + assert_eq!(stats.rchar_bytes, Some(11)); + assert_eq!(stats.wchar_bytes, Some(22)); + assert_eq!(stats.syscall_read_total, Some(33)); + assert_eq!(stats.syscall_write_total, Some(44)); + assert_eq!(stats.read_bytes, Some(55)); + assert_eq!(stats.write_bytes, Some(66)); + } + } +} + +#[cfg(target_os = "windows")] +mod platform { + use super::*; + + pub(super) fn snapshot() -> ProcessPlatformSnapshot { + let pid = std::process::id(); + let script = format!( + "Get-CimInstance Win32_Process -Filter \"ProcessId = {pid}\" | \ + Format-List -Property ReadOperationCount,WriteOperationCount,PeakVirtualSize" + ); + + let output = run_command("powershell", &["-NoProfile", "-Command", &script]); + ProcessPlatformSnapshot { + syscall_read_total: output.as_deref().and_then(|v| parse_kv_u64(v, "ReadOperationCount")), + syscall_write_total: output.as_deref().and_then(|v| parse_kv_u64(v, "WriteOperationCount")), + virtual_memory_max_bytes: output.as_deref().and_then(|v| parse_kv_u64(v, "PeakVirtualSize")), + ..Default::default() + } + } +} + +#[cfg(target_os = "macos")] +mod platform { + use super::*; + + pub(super) fn snapshot() -> ProcessPlatformSnapshot { + let pid = std::process::id().to_string(); + let output = run_command("ps", &["-o", "inblock=", "-o", "oublock=", "-o", "vsz=", "-p", &pid]); + parse_ps_stats(output.as_deref()) + } + + fn parse_ps_stats(output: Option<&str>) -> ProcessPlatformSnapshot { + let Some(output) = output else { + return ProcessPlatformSnapshot::default(); + }; + + let Some(line) = output.lines().find(|line| !line.trim().is_empty()) else { + return ProcessPlatformSnapshot::default(); + }; + + let mut parts = line.split_whitespace(); + let inblock = parts.next().and_then(|v| v.parse::().ok()); + let oublock = parts.next().and_then(|v| v.parse::().ok()); + let vsz_kb = parts.next().and_then(|v| v.parse::().ok()); + + ProcessPlatformSnapshot { + syscall_read_total: inblock, + syscall_write_total: oublock, + virtual_memory_max_bytes: vsz_kb.map(|v| v.saturating_mul(1024)), + ..Default::default() + } + } + + #[cfg(test)] + mod tests { + use super::*; + + #[test] + fn parse_ps_line() { + let snapshot = parse_ps_stats(Some("12 34 5678")); + assert_eq!(snapshot.syscall_read_total, Some(12)); + assert_eq!(snapshot.syscall_write_total, Some(34)); + assert_eq!(snapshot.virtual_memory_max_bytes, Some(5678 * 1024)); + } + } +} + +#[cfg(any(target_os = "freebsd", target_os = "openbsd", target_os = "netbsd", target_os = "dragonfly"))] +mod platform { + use super::*; + + pub(super) fn snapshot() -> ProcessPlatformSnapshot { + let pid = std::process::id().to_string(); + let output = run_command("ps", &["-o", "inblk=", "-o", "oublk=", "-o", "vsz=", "-p", &pid]); + parse_ps_stats(output.as_deref()) + } + + fn parse_ps_stats(output: Option<&str>) -> ProcessPlatformSnapshot { + let Some(output) = output else { + return ProcessPlatformSnapshot::default(); + }; + + let Some(line) = output.lines().find(|line| !line.trim().is_empty()) else { + return ProcessPlatformSnapshot::default(); + }; + + let mut parts = line.split_whitespace(); + let inblock = parts.next().and_then(|v| v.parse::().ok()); + let oublock = parts.next().and_then(|v| v.parse::().ok()); + let vsz_kb = parts.next().and_then(|v| v.parse::().ok()); + + ProcessPlatformSnapshot { + syscall_read_total: inblock, + syscall_write_total: oublock, + virtual_memory_max_bytes: vsz_kb.map(|v| v.saturating_mul(1024)), + ..Default::default() + } + } +} + +#[cfg(not(any( + target_os = "linux", + target_os = "windows", + target_os = "macos", + target_os = "freebsd", + target_os = "openbsd", + target_os = "netbsd", + target_os = "dragonfly" +)))] +mod platform { + use super::*; + + pub(super) fn snapshot() -> ProcessPlatformSnapshot { + ProcessPlatformSnapshot::default() + } +} + +#[cfg(test)] +pub fn reset_process_lock_counts() { + READ_LOCKS_HELD.store(0, Ordering::Relaxed); + WRITE_LOCKS_HELD.store(0, Ordering::Relaxed); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn held_lock_counters_round_trip() { + reset_process_lock_counts(); + + record_read_lock_held_acquire(); + record_read_lock_held_acquire(); + record_write_lock_held_acquire(); + + assert_eq!( + snapshot_process_lock_counts(), + ProcessLockSnapshot { + read_locks_held: 2, + write_locks_held: 1, + } + ); + + record_read_lock_held_release(); + record_write_lock_held_release(); + record_write_lock_held_release(); + + assert_eq!( + snapshot_process_lock_counts(), + ProcessLockSnapshot { + read_locks_held: 1, + write_locks_held: 0, + } + ); + } + + #[test] + fn parse_kv_u64_round_trip() { + let content = "syscr: 123\nsyscw: 456\n"; + assert_eq!(parse_kv_u64(content, "syscr"), Some(123)); + assert_eq!(parse_kv_u64(content, "syscw"), Some(456)); + assert_eq!(parse_kv_u64(content, "missing"), None); + } +} diff --git a/crates/io-metrics/src/sampler/mod.rs b/crates/io-metrics/src/sampler/mod.rs new file mode 100644 index 000000000..f793eb1fd --- /dev/null +++ b/crates/io-metrics/src/sampler/mod.rs @@ -0,0 +1,22 @@ +// 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. + +pub mod process; +pub mod system; + +pub use process::{ + ProcessResourceSnapshot, ProcessStatusSnapshot, ProcessSystemSnapshot, snapshot_process_resource, + snapshot_process_resource_and_system, snapshot_process_system, +}; +pub use system::snapshot_process_platform; diff --git a/crates/io-metrics/src/sampler/process.rs b/crates/io-metrics/src/sampler/process.rs new file mode 100644 index 000000000..cfc0fa8c9 --- /dev/null +++ b/crates/io-metrics/src/sampler/process.rs @@ -0,0 +1,164 @@ +// 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. + +use std::sync::OnceLock; +use std::time::Instant; +use sysinfo::{Pid, ProcessRefreshKind, ProcessStatus, ProcessesToUpdate, System}; + +/// Process start time used for computing process uptime. +static PROCESS_START: OnceLock = OnceLock::new(); + +#[inline] +fn get_process_start() -> &'static Instant { + PROCESS_START.get_or_init(Instant::now) +} + +#[derive(Debug, Clone, Copy, Default, PartialEq)] +pub struct ProcessResourceSnapshot { + pub cpu_percent: f64, + pub memory_bytes: u64, + pub uptime_seconds: u64, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum ProcessStatusSnapshot { + Running = 0, + Sleeping = 1, + Zombie = 2, + #[default] + Other = 3, +} + +impl From for ProcessStatusSnapshot { + fn from(status: ProcessStatus) -> Self { + match status { + ProcessStatus::Run => ProcessStatusSnapshot::Running, + ProcessStatus::Sleep => ProcessStatusSnapshot::Sleeping, + ProcessStatus::Zombie => ProcessStatusSnapshot::Zombie, + _ => ProcessStatusSnapshot::Other, + } + } +} + +#[derive(Debug, Clone, Copy, Default, PartialEq)] +pub struct ProcessSystemSnapshot { + pub locks_read_total: u64, + pub locks_write_total: u64, + pub cpu_total_seconds: f64, + pub go_routine_total: u64, + pub io_rchar_bytes: u64, + pub io_read_bytes: u64, + pub io_wchar_bytes: u64, + pub io_write_bytes: u64, + pub start_time_seconds: u64, + pub uptime_seconds: u64, + pub file_descriptor_limit_total: u64, + pub file_descriptor_open_total: u64, + pub syscall_read_total: u64, + pub syscall_write_total: u64, + pub resident_memory_bytes: u64, + pub virtual_memory_bytes: u64, + pub virtual_memory_max_bytes: u64, + pub status: ProcessStatusSnapshot, + pub status_value: i64, +} + +/// Collect resource-only process snapshot for the current process. +#[inline] +pub fn snapshot_process_resource() -> ProcessResourceSnapshot { + snapshot_process_resource_and_system().0 +} + +/// Collect system-level process snapshot for the current process. +#[inline] +pub fn snapshot_process_system() -> ProcessSystemSnapshot { + snapshot_process_resource_and_system().1 +} + +/// Collect both resource and system snapshots in one sysinfo refresh. +#[inline] +pub fn snapshot_process_resource_and_system() -> (ProcessResourceSnapshot, ProcessSystemSnapshot) { + let uptime_seconds = get_process_start().elapsed().as_secs(); + let platform_stats = crate::snapshot_process_platform_stats(); + let lock_snapshot = crate::snapshot_process_lock_counts(); + + let mut sys = System::new(); + let pid = Pid::from_u32(std::process::id()); + sys.refresh_processes_specifics(ProcessesToUpdate::Some(&[pid]), true, ProcessRefreshKind::everything()); + + if let Some(process) = sys.process(pid) { + let disk_usage = process.disk_usage(); + let status = ProcessStatusSnapshot::from(process.status()); + + let resource_stats = ProcessResourceSnapshot { + cpu_percent: process.cpu_usage() as f64, + memory_bytes: process.memory(), + uptime_seconds, + }; + + let process_stats = ProcessSystemSnapshot { + locks_read_total: lock_snapshot.read_locks_held, + locks_write_total: lock_snapshot.write_locks_held, + cpu_total_seconds: process.accumulated_cpu_time() as f64 / 1000.0, + file_descriptor_limit_total: process.open_files_limit().map_or(0, |value| value as u64), + file_descriptor_open_total: process.open_files().map_or(0, |value| value as u64), + go_routine_total: process.tasks().map_or(0, |tasks| tasks.len() as u64), + io_rchar_bytes: platform_stats.io_rchar_bytes.unwrap_or(disk_usage.total_read_bytes), + io_read_bytes: platform_stats.io_read_bytes.unwrap_or(disk_usage.total_read_bytes), + io_wchar_bytes: platform_stats.io_wchar_bytes.unwrap_or(disk_usage.total_written_bytes), + io_write_bytes: platform_stats.io_write_bytes.unwrap_or(disk_usage.total_written_bytes), + resident_memory_bytes: process.memory(), + start_time_seconds: process.start_time(), + status, + status_value: status as i64, + syscall_read_total: platform_stats.syscall_read_total.unwrap_or(0), + syscall_write_total: platform_stats.syscall_write_total.unwrap_or(0), + uptime_seconds, + virtual_memory_bytes: process.virtual_memory(), + virtual_memory_max_bytes: platform_stats.virtual_memory_max_bytes.unwrap_or(0), + }; + + (resource_stats, process_stats) + } else { + ( + ProcessResourceSnapshot { + uptime_seconds, + ..Default::default() + }, + ProcessSystemSnapshot { + uptime_seconds, + ..Default::default() + }, + ) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn process_status_snapshot_maps_sysinfo_values() { + assert_eq!(ProcessStatusSnapshot::from(ProcessStatus::Run), ProcessStatusSnapshot::Running); + assert_eq!(ProcessStatusSnapshot::from(ProcessStatus::Sleep), ProcessStatusSnapshot::Sleeping); + assert_eq!(ProcessStatusSnapshot::from(ProcessStatus::Zombie), ProcessStatusSnapshot::Zombie); + } + + #[test] + fn process_snapshots_are_collectable() { + let _ = snapshot_process_resource(); + let _ = snapshot_process_system(); + let _ = snapshot_process_resource_and_system(); + } +} diff --git a/crates/metrics/src/lib.rs b/crates/io-metrics/src/sampler/system.rs similarity index 72% rename from crates/metrics/src/lib.rs rename to crates/io-metrics/src/sampler/system.rs index 529c5bc40..e70864ef8 100644 --- a/crates/metrics/src/lib.rs +++ b/crates/io-metrics/src/sampler/system.rs @@ -12,12 +12,10 @@ // See the License for the specific language governing permissions and // limitations under the License. -pub mod collectors; -pub mod constants; -pub mod format; -mod global; -mod metrics_type; +use crate::ProcessPlatformSnapshot; -pub use format::report_metrics; -pub use global::init_metrics_system; -pub use metrics_type::*; +/// Collect process platform-specific I/O and virtual memory counters. +#[inline] +pub fn snapshot_process_platform() -> ProcessPlatformSnapshot { + crate::snapshot_process_platform_stats() +} diff --git a/crates/lock/Cargo.toml b/crates/lock/Cargo.toml index 6e1e9eb05..1ecfb0980 100644 --- a/crates/lock/Cargo.toml +++ b/crates/lock/Cargo.toml @@ -29,6 +29,7 @@ documentation = "https://docs.rs/rustfs-lock/latest/rustfs_lock/" workspace = true [dependencies] +rustfs-io-metrics = { workspace = true } rustfs-utils = { workspace = true } async-trait.workspace = true futures.workspace = true diff --git a/crates/lock/src/distributed_lock.rs b/crates/lock/src/distributed_lock.rs index c98a172be..3849106bc 100644 --- a/crates/lock/src/distributed_lock.rs +++ b/crates/lock/src/distributed_lock.rs @@ -19,6 +19,9 @@ use crate::{ types::{LockId, LockInfo, LockRequest, LockResponse, LockStatus, LockType}, }; use futures::future::join_all; +use rustfs_io_metrics::{ + record_read_lock_held_acquire, record_read_lock_held_release, record_write_lock_held_acquire, record_write_lock_held_release, +}; use std::sync::{Arc, LazyLock}; use std::time::Duration; use tokio::sync::mpsc; @@ -82,6 +85,7 @@ pub struct DistributedLockGuard { /// All underlying (LockId, client) entries that should be released when the /// guard is dropped. entries: Vec<(LockId, Arc)>, + lock_type: LockType, /// If true, Drop will not try to release (used if user manually released). disarmed: bool, } @@ -92,10 +96,12 @@ impl DistributedLockGuard { /// - `lock_id` is the id returned to the caller (`lock_id()`). /// - `entries` is the full list of underlying (LockId, client) pairs /// that should be released when this guard is dropped. - pub(crate) fn new(lock_id: LockId, entries: Vec<(LockId, Arc)>) -> Self { + pub(crate) fn new(lock_id: LockId, entries: Vec<(LockId, Arc)>, lock_type: LockType) -> Self { + record_lock_held_acquire(lock_type); Self { lock_id, entries, + lock_type, disarmed: false, } } @@ -108,6 +114,9 @@ impl DistributedLockGuard { /// Manually disarm the guard so dropping it won't release the lock. /// Call this if you explicitly released the lock elsewhere. pub fn disarm(&mut self) { + if !self.disarmed { + record_lock_held_release(self.lock_type); + } self.disarmed = true; } @@ -156,6 +165,7 @@ impl DistributedLockGuard { // Disarm to prevent double-release on drop self.disarmed = true; + record_lock_held_release(self.lock_type); success } } @@ -248,7 +258,7 @@ impl DistributedLock { .map(|info| info.id.clone()) .unwrap_or_else(|| LockId::new_unique(&request.resource)); - Ok(Some(DistributedLockGuard::new(aggregate_lock_id, individual_locks))) + Ok(Some(DistributedLockGuard::new(aggregate_lock_id, individual_locks, request.lock_type))) } else { // Check if it's a timeout or quorum failure if let Some(error_msg) = &resp.error { @@ -523,3 +533,19 @@ impl DistributedLock { Ok((resp, individual_locks)) } } + +#[inline(always)] +fn record_lock_held_acquire(lock_type: LockType) { + match lock_type { + LockType::Shared => record_read_lock_held_acquire(), + LockType::Exclusive => record_write_lock_held_acquire(), + } +} + +#[inline(always)] +fn record_lock_held_release(lock_type: LockType) { + match lock_type { + LockType::Shared => record_read_lock_held_release(), + LockType::Exclusive => record_write_lock_held_release(), + } +} diff --git a/crates/lock/src/fast_lock/guard.rs b/crates/lock/src/fast_lock/guard.rs index c1fea6e18..614d066d1 100644 --- a/crates/lock/src/fast_lock/guard.rs +++ b/crates/lock/src/fast_lock/guard.rs @@ -16,6 +16,9 @@ use crate::fast_lock::{ shard::LockShard, types::{LockMode, ObjectKey}, }; +use rustfs_io_metrics::{ + record_read_lock_held_acquire, record_read_lock_held_release, record_write_lock_held_acquire, record_write_lock_held_release, +}; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; @@ -40,6 +43,7 @@ pub struct FastLockGuard { impl FastLockGuard { pub(crate) fn new(key: ObjectKey, mode: LockMode, owner: Arc, shard: Arc) -> Self { let guard_id = GUARD_ID_COUNTER.fetch_add(1, Ordering::Relaxed); + record_lock_held_acquire(mode); Self { key, mode, @@ -102,6 +106,7 @@ impl FastLockGuard { let success = shard.release_lock_with_guard(&self.key, &self.owner, self.mode, self.guard_id); if success { self.released = true; + record_lock_held_release(self.mode); // Unregister the guard after successful release shard.unregister_guard(self.guard_id); } @@ -158,6 +163,9 @@ impl Drop for FastLockGuard { self.guard_id ); } + if success { + record_lock_held_release(self.mode); + } // Always unregister the guard to prevent leaks, regardless of release success shard.unregister_guard(self.guard_id); } else { @@ -168,6 +176,22 @@ impl Drop for FastLockGuard { } } +#[inline(always)] +fn record_lock_held_acquire(mode: LockMode) { + match mode { + LockMode::Shared => record_read_lock_held_acquire(), + LockMode::Exclusive => record_write_lock_held_acquire(), + } +} + +#[inline(always)] +fn record_lock_held_release(mode: LockMode) { + match mode { + LockMode::Shared => record_read_lock_held_release(), + LockMode::Exclusive => record_write_lock_held_release(), + } +} + impl std::fmt::Debug for FastLockGuard { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("FastLockGuard") diff --git a/crates/metrics/Cargo.toml b/crates/metrics/Cargo.toml deleted file mode 100644 index 06e71b1c4..000000000 --- a/crates/metrics/Cargo.toml +++ /dev/null @@ -1,46 +0,0 @@ -# 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. - -[package] -name = "rustfs-metrics" -edition.workspace = true -license.workspace = true -repository.workspace = true -rust-version.workspace = true -version.workspace = true -homepage.workspace = true -description.workspace = true -keywords.workspace = true -categories.workspace = true -authors.workspace = true - -[features] -default = [] -gpu = ["dep:nvml-wrapper"] -full = ["gpu"] - -[dependencies] -rustfs-config = { workspace = true } -rustfs-ecstore = { workspace = true } -rustfs-utils = { workspace = true } -metrics = { workspace = true } -sysinfo = { workspace = true } -tokio = { workspace = true, features = ["rt-multi-thread", "rt", "time", "macros"] } -tokio-util = { workspace = true } -tracing = { workspace = true } -thiserror = { workspace = true } -nvml-wrapper = { workspace = true, optional = true } - -[lints] -workspace = true diff --git a/crates/metrics/src/collectors/logger_webhook.rs b/crates/metrics/src/collectors/logger_webhook.rs deleted file mode 100644 index 48c6b37a8..000000000 --- a/crates/metrics/src/collectors/logger_webhook.rs +++ /dev/null @@ -1,122 +0,0 @@ -// 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. - -#![allow(dead_code)] - -//! Logger webhook metrics collector. -//! -//! Collects webhook metrics including failed messages, queue length, -//! and total messages per webhook target. - -use crate::format::PrometheusMetric; -use crate::metrics_type::logger_webhook::{ - ENDPOINT_LABEL, NAME_LABEL, WEBHOOK_FAILED_MESSAGES_MD, WEBHOOK_QUEUE_LENGTH_MD, WEBHOOK_TOTAL_MESSAGES_MD, -}; -use std::borrow::Cow; - -/// Webhook target statistics. -#[derive(Debug, Clone, Default)] -pub struct WebhookTargetStats { - /// Webhook name - pub name: String, - /// Webhook endpoint URL - pub endpoint: String, - /// Number of failed messages - pub failed_messages: u64, - /// Number of messages in queue - pub queue_length: u64, - /// Total number of messages sent - pub total_messages: u64, -} - -/// Collects webhook metrics from the given stats. -/// -/// Uses the metric descriptors from `metrics_type::logger_webhook` module. -/// Returns a vector of Prometheus metrics for webhook statistics. -pub fn collect_webhook_metrics(stats: &[WebhookTargetStats]) -> Vec { - let mut metrics = Vec::with_capacity(stats.len() * 3); - - for stat in stats { - let name_label: Cow<'static, str> = Cow::Owned(stat.name.clone()); - let endpoint_label: Cow<'static, str> = Cow::Owned(stat.endpoint.clone()); - - metrics.push( - PrometheusMetric::from_descriptor(&WEBHOOK_FAILED_MESSAGES_MD, stat.failed_messages as f64) - .with_label(NAME_LABEL, name_label.clone()) - .with_label(ENDPOINT_LABEL, endpoint_label.clone()), - ); - metrics.push( - PrometheusMetric::from_descriptor(&WEBHOOK_QUEUE_LENGTH_MD, stat.queue_length as f64) - .with_label(NAME_LABEL, name_label.clone()) - .with_label(ENDPOINT_LABEL, endpoint_label.clone()), - ); - metrics.push( - PrometheusMetric::from_descriptor(&WEBHOOK_TOTAL_MESSAGES_MD, stat.total_messages as f64) - .with_label(NAME_LABEL, name_label.clone()) - .with_label(ENDPOINT_LABEL, endpoint_label.clone()), - ); - } - - metrics -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::format::report_metrics; - - #[test] - fn test_collect_webhook_metrics() { - let stats = vec![WebhookTargetStats { - name: "alert-webhook".to_string(), - endpoint: "https://hooks.example.com/alert".to_string(), - failed_messages: 3, - queue_length: 15, - total_messages: 5000, - }]; - - let metrics = collect_webhook_metrics(&stats); - report_metrics(&metrics); - - assert_eq!(metrics.len(), 3); - - // Verify we have metrics with the expected values - let failed = metrics.iter().find(|m| m.value == 3.0); - assert!(failed.is_some()); - - // Verify labels - let total = metrics.iter().find(|m| m.value == 5000.0); - assert!(total.is_some()); - let total_metric = total.unwrap(); - assert!( - total_metric - .labels - .iter() - .any(|(k, v)| *k == NAME_LABEL && v == "alert-webhook") - ); - assert!( - total_metric - .labels - .iter() - .any(|(k, v)| *k == ENDPOINT_LABEL && v == "https://hooks.example.com/alert") - ); - } - - #[test] - fn test_collect_webhook_metrics_empty() { - let stats: Vec = vec![]; - let metrics = collect_webhook_metrics(&stats); - assert!(metrics.is_empty()); - } -} diff --git a/crates/metrics/src/global.rs b/crates/metrics/src/global.rs deleted file mode 100644 index f37378d0c..000000000 --- a/crates/metrics/src/global.rs +++ /dev/null @@ -1,40 +0,0 @@ -// 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. - -use tokio_util::sync::CancellationToken; -use tracing::info; - -/// Initializes the global metrics system. This should be called once at the start of the application. -/// The provided `CancellationToken` will be used to gracefully shut down the metrics system when needed. -/// -/// # Arguments -/// * `token` - A `CancellationToken` that can be used to signal the metrics system to shut down gracefully. -/// -/// # Example -/// ```ignore -/// use tokio_util::sync::CancellationToken; -/// use rustfs_metrics::init_metrics_system; -/// -/// let token = CancellationToken::new(); -/// init_metrics_system(token.clone()); -/// -/// // Later, when you want to shut down the metrics system: -/// token.cancel(); -/// ``` -/// Note: This function should only be called once during the application's lifecycle. Calling it multiple times may lead to unexpected behavior. -pub fn init_metrics_system(token: CancellationToken) { - info!("init metrics system start"); - crate::collectors::init_metrics_collectors(token); - info!("init metrics system done"); -} diff --git a/crates/metrics/src/metrics_type/logger_webhook.rs b/crates/metrics/src/metrics_type/logger_webhook.rs deleted file mode 100644 index dfe41b179..000000000 --- a/crates/metrics/src/metrics_type/logger_webhook.rs +++ /dev/null @@ -1,54 +0,0 @@ -// 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. - -#![allow(dead_code)] - -use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems}; -use std::sync::LazyLock; - -/// Define label constants for webhook metrics -/// name label -pub const NAME_LABEL: &str = "name"; -/// endpoint label -pub const ENDPOINT_LABEL: &str = "endpoint"; - -// The label used by all webhook metrics -const ALL_WEBHOOK_LABELS: [&str; 2] = [NAME_LABEL, ENDPOINT_LABEL]; - -pub static WEBHOOK_FAILED_MESSAGES_MD: LazyLock = LazyLock::new(|| { - new_counter_md( - MetricName::WebhookFailedMessages, - "Number of messages that failed to send", - &ALL_WEBHOOK_LABELS[..], - subsystems::LOGGER_WEBHOOK, - ) -}); - -pub static WEBHOOK_QUEUE_LENGTH_MD: LazyLock = LazyLock::new(|| { - new_gauge_md( - MetricName::WebhookQueueLength, - "Webhook queue length", - &ALL_WEBHOOK_LABELS[..], - subsystems::LOGGER_WEBHOOK, - ) -}); - -pub static WEBHOOK_TOTAL_MESSAGES_MD: LazyLock = LazyLock::new(|| { - new_counter_md( - MetricName::WebhookTotalMessages, - "Total number of messages sent to this target", - &ALL_WEBHOOK_LABELS[..], - subsystems::LOGGER_WEBHOOK, - ) -}); diff --git a/crates/notify/src/global.rs b/crates/notify/src/global.rs index 5280fd0d0..a0c716f7e 100644 --- a/crates/notify/src/global.rs +++ b/crates/notify/src/global.rs @@ -12,7 +12,10 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::{BucketNotificationConfig, Event, EventArgs, LifecycleError, NotificationError, NotificationSystem}; +use crate::{ + BucketNotificationConfig, Event, EventArgs, LifecycleError, NotificationError, NotificationMetricSnapshot, + NotificationSystem, NotificationTargetMetricSnapshot, +}; use rustfs_ecstore::config::Config; use rustfs_s3_common::EventName; use rustfs_targets::arn::TargetID; @@ -41,6 +44,23 @@ pub fn notification_system() -> Option> { NOTIFICATION_SYSTEM.get().cloned() } +/// Returns aggregate notification delivery metrics for Prometheus collection. +pub fn notification_metrics_snapshot() -> NotificationMetricSnapshot { + NOTIFICATION_SYSTEM + .get() + .map(|system| system.snapshot_metrics()) + .unwrap_or_default() +} + +/// Returns per-target notification delivery metrics for Prometheus collection. +pub async fn notification_target_metrics() -> Vec { + if let Some(system) = notification_system() { + system.snapshot_target_metrics().await + } else { + Vec::new() + } +} + /// Check if the notification system has been initialized. pub fn is_notification_system_initialized() -> bool { NOTIFICATION_SYSTEM.get().is_some() diff --git a/crates/notify/src/integration.rs b/crates/notify/src/integration.rs index 4ac5087e7..12fab50f4 100644 --- a/crates/notify/src/integration.rs +++ b/crates/notify/src/integration.rs @@ -106,10 +106,29 @@ pub struct NotificationMetrics { processed_events: AtomicUsize, /// Number of events that failed to handle failed_events: AtomicUsize, + /// Number of dispatch attempts skipped before delivery + skipped_events: AtomicUsize, /// System startup time start_time: Instant, } +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct NotificationMetricSnapshot { + pub current_send_in_progress: u64, + pub events_errors_total: u64, + pub events_sent_total: u64, + pub events_skipped_total: u64, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct NotificationTargetMetricSnapshot { + pub failed_messages: u64, + pub queue_length: u64, + pub target_id: String, + pub target_type: String, + pub total_messages: u64, +} + impl Default for NotificationMetrics { fn default() -> Self { Self::new() @@ -122,6 +141,7 @@ impl NotificationMetrics { processing_events: AtomicUsize::new(0), processed_events: AtomicUsize::new(0), failed_events: AtomicUsize::new(0), + skipped_events: AtomicUsize::new(0), start_time: Instant::now(), } } @@ -136,11 +156,19 @@ impl NotificationMetrics { self.processed_events.fetch_add(1, Ordering::Relaxed); } + pub fn decrement_processing(&self) { + self.processing_events.fetch_sub(1, Ordering::Relaxed); + } + pub fn increment_failed(&self) { self.processing_events.fetch_sub(1, Ordering::Relaxed); self.failed_events.fetch_add(1, Ordering::Relaxed); } + pub fn increment_skipped(&self) { + self.skipped_events.fetch_add(1, Ordering::Relaxed); + } + // Provide public methods to get count pub fn processing_count(&self) -> usize { self.processing_events.load(Ordering::Relaxed) @@ -154,9 +182,22 @@ impl NotificationMetrics { self.failed_events.load(Ordering::Relaxed) } + pub fn skipped_count(&self) -> usize { + self.skipped_events.load(Ordering::Relaxed) + } + pub fn uptime(&self) -> Duration { self.start_time.elapsed() } + + pub fn snapshot(&self) -> NotificationMetricSnapshot { + NotificationMetricSnapshot { + current_send_in_progress: self.processing_count() as u64, + events_errors_total: self.failed_count() as u64, + events_sent_total: self.processed_count() as u64, + events_skipped_total: self.skipped_count() as u64, + } + } } /// The notification system that integrates all components @@ -187,14 +228,15 @@ impl NotificationSystem { let concurrency_limiter = rustfs_utils::get_env_usize(ENV_NOTIFY_TARGET_STREAM_CONCURRENCY, DEFAULT_NOTIFY_TARGET_STREAM_CONCURRENCY); let (live_event_sender, _) = broadcast::channel(1024); + let metrics = Arc::new(NotificationMetrics::new()); NotificationSystem { subscriber_view: NotificationSystemSubscriberView::new(), - notifier: Arc::new(EventNotifier::new()), + notifier: Arc::new(EventNotifier::new(metrics.clone())), registry: Arc::new(TargetRegistry::new()), config: Arc::new(RwLock::new(config)), stream_cancellers: Arc::new(RwLock::new(HashMap::new())), concurrency_limiter: Arc::new(Semaphore::new(concurrency_limiter)), // Limit the maximum number of concurrent processing events to 20 - metrics: Arc::new(NotificationMetrics::new()), + metrics, live_event_sender, live_event_history: Arc::new(RwLock::new(LiveEventHistory::default())), } @@ -585,10 +627,35 @@ impl NotificationSystem { status.insert("processing_events".to_string(), self.metrics.processing_count().to_string()); status.insert("processed_events".to_string(), self.metrics.processed_count().to_string()); status.insert("failed_events".to_string(), self.metrics.failed_count().to_string()); + status.insert("skipped_events".to_string(), self.metrics.skipped_count().to_string()); status } + pub fn snapshot_metrics(&self) -> NotificationMetricSnapshot { + self.metrics.snapshot() + } + + pub async fn snapshot_target_metrics(&self) -> Vec { + let targets = self.notifier.target_list().read().await.values(); + let mut snapshots = Vec::with_capacity(targets.len()); + + for target in targets { + let delivery = target.delivery_snapshot(); + let target_id = target.id(); + snapshots.push(NotificationTargetMetricSnapshot { + failed_messages: delivery.failed_messages, + queue_length: delivery.queue_length, + target_id: target_id.to_string(), + target_type: target_id.name, + total_messages: delivery.total_messages, + }); + } + + snapshots.sort_by(|a, b| a.target_id.cmp(&b.target_id)); + snapshots + } + // Add a method to shut down the system pub async fn shutdown(&self) { info!("Turn off the notification system"); diff --git a/crates/notify/src/lib.rs b/crates/notify/src/lib.rs index 4181e4d07..267367230 100644 --- a/crates/notify/src/lib.rs +++ b/crates/notify/src/lib.rs @@ -31,6 +31,9 @@ pub mod stream; pub use error::{LifecycleError, NotificationError}; pub use event::{Event, EventArgs, EventArgsBuilder}; -pub use global::{initialize, is_notification_system_initialized, notification_system, notifier_global}; -pub use integration::NotificationSystem; +pub use global::{ + initialize, is_notification_system_initialized, notification_metrics_snapshot, notification_system, + notification_target_metrics, notifier_global, +}; +pub use integration::{NotificationMetricSnapshot, NotificationSystem, NotificationTargetMetricSnapshot}; pub use rules::BucketNotificationConfig; diff --git a/crates/notify/src/notifier.rs b/crates/notify/src/notifier.rs index 4d6fe40c3..aac12d1c9 100644 --- a/crates/notify/src/notifier.rs +++ b/crates/notify/src/notifier.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::{error::NotificationError, event::Event, rules::RulesMap}; +use crate::{error::NotificationError, event::Event, integration::NotificationMetrics, rules::RulesMap}; use hashbrown::HashMap; use rustfs_config::notify::{DEFAULT_NOTIFY_SEND_CONCURRENCY, ENV_NOTIFY_SEND_CONCURRENCY}; use rustfs_s3_common::EventName; @@ -26,6 +26,7 @@ use tracing::{debug, error, info, instrument, warn}; /// Manages event notification to targets based on rules pub struct EventNotifier { + metrics: Arc, target_list: Arc>, bucket_rules_map: Arc>, send_limiter: Arc, @@ -33,7 +34,7 @@ pub struct EventNotifier { impl Default for EventNotifier { fn default() -> Self { - Self::new() + Self::new(Arc::new(NotificationMetrics::new())) } } @@ -42,9 +43,10 @@ impl EventNotifier { /// /// # Returns /// Returns a new instance of EventNotifier. - pub fn new() -> Self { + pub fn new(metrics: Arc) -> Self { let max_inflight = rustfs_utils::get_env_usize(ENV_NOTIFY_SEND_CONCURRENCY, DEFAULT_NOTIFY_SEND_CONCURRENCY); EventNotifier { + metrics, target_list: Arc::new(RwLock::new(TargetList::new())), bucket_rules_map: Arc::new(AsyncShardedHashMap::new(0)), send_limiter: Arc::new(Semaphore::new(max_inflight)), @@ -182,12 +184,14 @@ impl EventNotifier { let Some(rules) = self.bucket_rules_map.get(bucket_name).await else { debug!("No rules found for bucket: {}", bucket_name); + self.metrics.increment_skipped(); return; }; let target_ids = rules.match_rules(event_name, object_key); if target_ids.is_empty() { debug!("No matching targets for event in bucket: {}", bucket_name); + self.metrics.increment_skipped(); return; } let target_ids_len = target_ids.len(); @@ -207,7 +211,9 @@ impl EventNotifier { continue; } let limiter = self.send_limiter.clone(); + let metrics = self.metrics.clone(); let event_clone = event.clone(); + let is_deferred = target_for_task.store().is_some(); let target_name_for_task = target_for_task.name(); // Get the name before generating the task debug!("Preparing to send event to target: {}", target_name_for_task); // Use cloned data in closures to avoid borrowing conflicts @@ -219,22 +225,31 @@ impl EventNotifier { data: event_clone.as_ref().clone(), }); let handle = tokio::spawn(async move { + metrics.increment_processing(); let _permit = match limiter.acquire_owned().await { Ok(p) => p, Err(e) => { error!("Failed to acquire send permit for target {}: {}", target_name_for_task, e); + metrics.increment_failed(); return; } }; if let Err(e) = target_for_task.save(entity_target.clone()).await { + metrics.increment_failed(); error!("Failed to send event to target {}: {}", target_name_for_task, e); } else { + if is_deferred { + metrics.decrement_processing(); + } else { + metrics.increment_processed(); + } debug!("Successfully saved event to target {}", target_name_for_task); } }); handles.push(handle); } else { warn!("Target ID {:?} found in rules but not in target list.", target_id); + self.metrics.increment_skipped(); } } // target_list is automatically released here @@ -470,7 +485,7 @@ mod tests { #[tokio::test] async fn test_send_event_skips_disabled_target() { - let notifier = EventNotifier::new(); + let notifier = EventNotifier::new(Arc::new(NotificationMetrics::new())); let enabled_target = TestTarget::new("enabled-target", "webhook", true); let disabled_target = TestTarget::new("disabled-target", "webhook", false); diff --git a/crates/notify/src/stream.rs b/crates/notify/src/stream.rs index 597d824e1..19f5ade2e 100644 --- a/crates/notify/src/stream.rs +++ b/crates/notify/src/stream.rs @@ -311,8 +311,14 @@ async fn process_batch( let backoff = 1u32 << retry_count as u32; tokio::time::sleep(base_delay * backoff + jitter).await; } + TargetError::Dropped(reason) => { + warn!("Dropped queued payload for target {}: {}", target.name(), reason); + metrics.increment_failed(); + break; + } _ => { error!("Permanent error for target {}: {}", target.name(), e); + target.record_final_failure(); metrics.increment_failed(); break; } @@ -322,6 +328,7 @@ async fn process_batch( if retry_count >= max_retries && !success { warn!("Max retries exceeded for event {}, target: {}, skipping", key.to_string(), target.name()); + target.record_final_failure(); metrics.increment_failed(); } } diff --git a/crates/obs/Cargo.toml b/crates/obs/Cargo.toml index 7a00bc879..c4d5a02f2 100644 --- a/crates/obs/Cargo.toml +++ b/crates/obs/Cargo.toml @@ -25,11 +25,19 @@ keywords = ["observability", "metrics", "logging", "tracing", "RustFS"] categories = ["web-programming", "development-tools::profiling", "asynchronous", "api-bindings", "development-tools::debugging"] documentation = "https://docs.rs/rustfs-obs/latest/rustfs_obs/" +[features] +default = [] +gpu = ["dep:nvml-wrapper"] + [lints] workspace = true [dependencies] +rustfs-audit = { workspace = true } rustfs-config = { workspace = true, features = ["constants", "observability"] } +rustfs-ecstore = { workspace = true } +rustfs-io-metrics = { workspace = true } +rustfs-notify = { workspace = true } rustfs-utils = { workspace = true, features = ["ip"] } flate2 = { workspace = true } glob = { workspace = true } @@ -52,9 +60,12 @@ tracing-error = { workspace = true } tracing-opentelemetry = { workspace = true } tracing-subscriber = { workspace = true, features = ["registry", "std", "fmt", "env-filter", "tracing-log", "time", "local-time", "json"] } tokio = { workspace = true, features = ["sync", "fs", "rt-multi-thread", "rt", "time", "macros"] } +tokio-util = { workspace = true } dial9-tokio-telemetry = { workspace = true } thiserror = { workspace = true } zstd = { workspace = true, features = ["zstdmt"] } +sysinfo = { workspace = true } +nvml-wrapper = { workspace = true, optional = true } [target.'cfg(any(target_os = "linux", target_os = "macos"))'.dependencies] pyroscope = { workspace = true, features = ["backend-pprof-rs"] } diff --git a/crates/obs/src/global.rs b/crates/obs/src/global.rs index c71fdade9..462da96a6 100644 --- a/crates/obs/src/global.rs +++ b/crates/obs/src/global.rs @@ -116,8 +116,7 @@ pub async fn init_obs(endpoint: Option) -> Result Result { let otel_guard = init_telemetry(config)?; - // Note: System monitoring has been migrated to rustfs-metrics - // Use rustfs_metrics::init_metrics_collectors() for system metrics + // Metrics runtime scheduling is exposed by rustfs_obs::init_metrics_runtime(). Ok(otel_guard) } diff --git a/crates/obs/src/lib.rs b/crates/obs/src/lib.rs index 521378019..e7d8a89f5 100644 --- a/crates/obs/src/lib.rs +++ b/crates/obs/src/lib.rs @@ -47,28 +47,30 @@ //! # } //! ``` //! -//! ## System Monitoring Migration +//! ## Metrics Runtime //! -//! The system monitoring functionality has been migrated to `rustfs-metrics`. -//! Use `rustfs_metrics::init_metrics_collectors()` for system metrics collection. +//! Start metrics scheduling with `rustfs_obs::init_metrics_runtime()`. //! //! ```ignore //! use tokio_util::sync::CancellationToken; -//! use rustfs_metrics::init_metrics_collectors; +//! use rustfs_obs::init_metrics_runtime; //! //! let token = CancellationToken::new(); -//! init_metrics_collectors(token.clone()); +//! init_metrics_runtime(token.clone()); //! ``` mod cleaner; mod config; mod error; mod global; +pub mod metrics; mod telemetry; pub use cleaner::*; pub use config::*; pub use error::*; pub use global::*; +pub use metrics::schema::*; +pub use metrics::{init_metrics_collectors, init_metrics_runtime}; pub use telemetry::{OtelGuard, Recorder}; // Dial9 Tokio runtime telemetry diff --git a/crates/metrics/src/collectors/audit.rs b/crates/obs/src/metrics/collectors/audit.rs similarity index 97% rename from crates/metrics/src/collectors/audit.rs rename to crates/obs/src/metrics/collectors/audit.rs index 9eda890f1..13700b1e1 100644 --- a/crates/metrics/src/collectors/audit.rs +++ b/crates/obs/src/metrics/collectors/audit.rs @@ -22,8 +22,8 @@ //! This collector reuses the metric descriptors defined in `metrics_type::audit` //! to avoid duplication of metric names, types, and help text. -use crate::format::PrometheusMetric; -use crate::metrics_type::audit::*; +use crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::audit::*; use std::borrow::Cow; /// Audit target statistics for metrics collection. diff --git a/crates/metrics/src/collectors/bucket.rs b/crates/obs/src/metrics/collectors/bucket.rs similarity index 97% rename from crates/metrics/src/collectors/bucket.rs rename to crates/obs/src/metrics/collectors/bucket.rs index ca421dc92..ab988bd21 100644 --- a/crates/metrics/src/collectors/bucket.rs +++ b/crates/obs/src/metrics/collectors/bucket.rs @@ -20,8 +20,8 @@ //! This collector reuses the metric descriptors defined in `metrics_type::node_bucket` //! to avoid duplication of metric names, types, and help text. -use crate::format::PrometheusMetric; -use crate::metrics_type::node_bucket::*; +use crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::node_bucket::*; use std::borrow::Cow; /// Bucket statistics for metrics collection. @@ -76,7 +76,7 @@ pub fn collect_bucket_metrics(buckets: &[BucketStats]) -> Vec #[cfg(test)] mod tests { use super::*; - use crate::format::report_metrics; + use crate::metrics::report::report_metrics; #[test] fn test_collect_bucket_metrics() { diff --git a/crates/metrics/src/collectors/bucket_replication.rs b/crates/obs/src/metrics/collectors/bucket_replication.rs similarity index 96% rename from crates/metrics/src/collectors/bucket_replication.rs rename to crates/obs/src/metrics/collectors/bucket_replication.rs index b0ce8e8a9..938f50982 100644 --- a/crates/metrics/src/collectors/bucket_replication.rs +++ b/crates/obs/src/metrics/collectors/bucket_replication.rs @@ -19,8 +19,8 @@ //! This collector reuses the metric descriptors defined in `metrics_type::bucket_replication` //! to avoid duplication of metric names, types, and help text. -use crate::format::PrometheusMetric; -use crate::metrics_type::bucket_replication::{BUCKET_REPL_BANDWIDTH_CURRENT_MD, BUCKET_REPL_BANDWIDTH_LIMIT_MD}; +use crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::bucket_replication::{BUCKET_REPL_BANDWIDTH_CURRENT_MD, BUCKET_REPL_BANDWIDTH_LIMIT_MD}; use std::borrow::Cow; /// Bucket replication bandwidth statistics for metrics collection. diff --git a/crates/metrics/src/collectors/cluster.rs b/crates/obs/src/metrics/collectors/cluster.rs similarity index 97% rename from crates/metrics/src/collectors/cluster.rs rename to crates/obs/src/metrics/collectors/cluster.rs index 6f512b28d..87a961d96 100644 --- a/crates/metrics/src/collectors/cluster.rs +++ b/crates/obs/src/metrics/collectors/cluster.rs @@ -20,8 +20,8 @@ //! This collector reuses the metric descriptors defined in `metrics_type::cluster` //! to avoid duplication of metric names, types, and help text. -use crate::format::PrometheusMetric; -use crate::metrics_type::cluster::*; +use crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::cluster::*; /// Cluster capacity and usage statistics for metrics collection. /// @@ -62,7 +62,7 @@ pub fn collect_cluster_metrics(stats: &ClusterStats) -> Vec { #[cfg(test)] mod tests { use super::*; - use crate::format::report_metrics; + use crate::metrics::report::report_metrics; #[test] fn test_collect_cluster_metrics() { diff --git a/crates/metrics/src/collectors/cluster_config.rs b/crates/obs/src/metrics/collectors/cluster_config.rs similarity index 96% rename from crates/metrics/src/collectors/cluster_config.rs rename to crates/obs/src/metrics/collectors/cluster_config.rs index cac2723fd..e079be17e 100644 --- a/crates/metrics/src/collectors/cluster_config.rs +++ b/crates/obs/src/metrics/collectors/cluster_config.rs @@ -22,8 +22,8 @@ //! This collector reuses the metric descriptors defined in `metrics_type::cluster_config` //! to avoid duplication of metric names, types, and help text. -use crate::format::PrometheusMetric; -use crate::metrics_type::cluster_config::*; +use crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::cluster_config::*; /// Cluster configuration statistics. #[derive(Debug, Clone, Default)] diff --git a/crates/metrics/src/collectors/cluster_erasure_set.rs b/crates/obs/src/metrics/collectors/cluster_erasure_set.rs similarity index 97% rename from crates/metrics/src/collectors/cluster_erasure_set.rs rename to crates/obs/src/metrics/collectors/cluster_erasure_set.rs index 4b12a461b..0a94b10a3 100644 --- a/crates/metrics/src/collectors/cluster_erasure_set.rs +++ b/crates/obs/src/metrics/collectors/cluster_erasure_set.rs @@ -19,8 +19,8 @@ //! Collects erasure coding set metrics including parity, quorum, //! drive counts, and health status. -use crate::format::PrometheusMetric; -use crate::metrics_type::cluster_erasure_set::*; +use crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::cluster_erasure_set::*; /// Erasure set statistics. #[derive(Debug, Clone, Default)] @@ -133,7 +133,7 @@ pub fn collect_erasure_set_metrics(stats: &[ErasureSetStats]) -> Vec Vec { #[cfg(test)] mod tests { use super::*; - use crate::format::report_metrics; + use crate::metrics::report::report_metrics; #[test] fn test_collect_iam_metrics() { diff --git a/crates/metrics/src/collectors/cluster_usage.rs b/crates/obs/src/metrics/collectors/cluster_usage.rs similarity index 98% rename from crates/metrics/src/collectors/cluster_usage.rs rename to crates/obs/src/metrics/collectors/cluster_usage.rs index 4c38f2a6b..a1259402b 100644 --- a/crates/metrics/src/collectors/cluster_usage.rs +++ b/crates/obs/src/metrics/collectors/cluster_usage.rs @@ -19,8 +19,8 @@ //! Collects cluster-wide and per-bucket usage metrics including //! object counts, sizes, versions, and distributions. -use crate::format::PrometheusMetric; -use crate::metrics_type::cluster_usage::*; +use crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::cluster_usage::*; /// Cluster-wide usage statistics. #[derive(Debug, Clone, Default)] @@ -148,7 +148,7 @@ pub fn collect_bucket_usage_metrics(stats: &[BucketUsageStats]) -> Vec Vec Vec { + if stats.is_empty() { + return Vec::new(); + } + + let mut metrics = Vec::with_capacity(stats.len() * 3); + for stat in stats { + let target_id: Cow<'static, str> = Cow::Owned(stat.target_id.clone()); + let target_type: Cow<'static, str> = Cow::Owned(stat.target_type.clone()); + + metrics.push( + PrometheusMetric::from_descriptor(&NOTIFICATION_TARGET_FAILED_MESSAGES_MD, stat.failed_messages as f64) + .with_label(TARGET_ID, target_id.clone()) + .with_label(TARGET_TYPE, target_type.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&NOTIFICATION_TARGET_QUEUE_LENGTH_MD, stat.queue_length as f64) + .with_label(TARGET_ID, target_id.clone()) + .with_label(TARGET_TYPE, target_type.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&NOTIFICATION_TARGET_TOTAL_MESSAGES_MD, stat.total_messages as f64) + .with_label(TARGET_ID, target_id) + .with_label(TARGET_TYPE, target_type), + ); + } + + metrics +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_collect_notification_target_metrics() { + let stats = vec![NotificationTargetStats { + failed_messages: 2, + queue_length: 4, + target_id: "primary:webhook".to_string(), + target_type: "webhook".to_string(), + total_messages: 42, + }]; + + let metrics = collect_notification_target_metrics(&stats); + + assert_eq!(metrics.len(), 3); + assert!(metrics.iter().any(|metric| { + metric.value == 42.0 + && metric + .labels + .iter() + .any(|(key, value)| *key == TARGET_ID && value == "primary:webhook") + && metric + .labels + .iter() + .any(|(key, value)| *key == TARGET_TYPE && value == "webhook") + })); + } +} diff --git a/crates/metrics/src/collectors/replication.rs b/crates/obs/src/metrics/collectors/replication.rs similarity index 96% rename from crates/metrics/src/collectors/replication.rs rename to crates/obs/src/metrics/collectors/replication.rs index 387025c83..c6a5c2878 100644 --- a/crates/metrics/src/collectors/replication.rs +++ b/crates/obs/src/metrics/collectors/replication.rs @@ -19,8 +19,8 @@ //! Collects cluster-wide replication metrics including queue stats, //! data transfer rates, and worker information. -use crate::format::PrometheusMetric; -use crate::metrics_type::replication::*; +use crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::replication::*; /// Replication statistics. #[derive(Debug, Clone, Default)] @@ -65,7 +65,7 @@ pub fn collect_replication_metrics(stats: &ReplicationStats) -> Vec Vec Vec #[cfg(test)] mod tests { use super::*; - use crate::format::report_metrics; + use crate::metrics::report::report_metrics; #[test] fn test_collect_resource_metrics() { diff --git a/crates/metrics/src/collectors/scanner.rs b/crates/obs/src/metrics/collectors/scanner.rs similarity index 96% rename from crates/metrics/src/collectors/scanner.rs rename to crates/obs/src/metrics/collectors/scanner.rs index 4b3bcf3d2..37af62689 100644 --- a/crates/metrics/src/collectors/scanner.rs +++ b/crates/obs/src/metrics/collectors/scanner.rs @@ -19,8 +19,8 @@ //! Collects background scanner metrics including bucket scans, //! directory scans, and object scans. -use crate::format::PrometheusMetric; -use crate::metrics_type::scanner::{ +use crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::scanner::{ SCANNER_BUCKET_SCANS_FINISHED_MD, SCANNER_BUCKET_SCANS_STARTED_MD, SCANNER_DIRECTORIES_SCANNED_MD, SCANNER_LAST_ACTIVITY_SECONDS_MD, SCANNER_OBJECTS_SCANNED_MD, SCANNER_VERSIONS_SCANNED_MD, }; @@ -60,7 +60,7 @@ pub fn collect_scanner_metrics(stats: &ScannerStats) -> Vec { #[cfg(test)] mod tests { use super::*; - use crate::format::report_metrics; + use crate::metrics::report::report_metrics; #[test] fn test_collect_scanner_metrics() { diff --git a/crates/metrics/src/collectors/system_cpu.rs b/crates/obs/src/metrics/collectors/system_cpu.rs similarity index 96% rename from crates/metrics/src/collectors/system_cpu.rs rename to crates/obs/src/metrics/collectors/system_cpu.rs index 07ea9cba9..10f682c44 100644 --- a/crates/metrics/src/collectors/system_cpu.rs +++ b/crates/obs/src/metrics/collectors/system_cpu.rs @@ -22,9 +22,9 @@ //! This module provides both system-level and process-level CPU metrics, //! with process-level metrics migrated from `rustfs-obs::system`. -use crate::format::PrometheusMetric; -use crate::metrics_type::system_cpu::*; -use crate::metrics_type::system_process::{PROCESS_CPU_USAGE_MD, PROCESS_CPU_UTILIZATION_MD}; +use crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::system_cpu::*; +use crate::metrics::schema::system_process::{PROCESS_CPU_USAGE_MD, PROCESS_CPU_UTILIZATION_MD}; use std::borrow::Cow; /// System CPU statistics. @@ -102,7 +102,7 @@ pub fn collect_process_cpu_metrics( #[cfg(test)] mod tests { use super::*; - use crate::format::report_metrics; + use crate::metrics::report::report_metrics; #[test] fn test_collect_cpu_metrics() { diff --git a/crates/metrics/src/collectors/system_drive.rs b/crates/obs/src/metrics/collectors/system_drive.rs similarity index 97% rename from crates/metrics/src/collectors/system_drive.rs rename to crates/obs/src/metrics/collectors/system_drive.rs index be4f71950..6d47f4dd1 100644 --- a/crates/metrics/src/collectors/system_drive.rs +++ b/crates/obs/src/metrics/collectors/system_drive.rs @@ -22,9 +22,9 @@ //! This module provides both system-level and process-level disk metrics, //! with process-level metrics migrated from `rustfs-obs::system`. -use crate::format::PrometheusMetric; -use crate::metrics_type::system_drive::*; -use crate::metrics_type::system_process::PROCESS_DISK_IO_MD; +use crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::system_drive::*; +use crate::metrics::schema::system_process::PROCESS_DISK_IO_MD; use std::borrow::Cow; /// Detailed drive statistics for a single drive. @@ -91,7 +91,7 @@ pub struct DriveCountStats { pub fn collect_drive_detailed_metrics(stats: &[DriveDetailedStats]) -> Vec { fn push_drive_metric( metrics: &mut Vec, - descriptor: &'static crate::MetricDescriptor, + descriptor: &'static crate::metrics::schema::MetricDescriptor, value: f64, server_label: &str, drive_label: &str, @@ -209,7 +209,7 @@ pub fn collect_process_disk_metrics( #[cfg(test)] mod tests { use super::*; - use crate::format::report_metrics; + use crate::metrics::report::report_metrics; #[test] fn test_collect_drive_detailed_metrics() { diff --git a/crates/metrics/src/collectors/system_gpu.rs b/crates/obs/src/metrics/collectors/system_gpu.rs similarity index 94% rename from crates/metrics/src/collectors/system_gpu.rs rename to crates/obs/src/metrics/collectors/system_gpu.rs index eca9a786b..183d48a1b 100644 --- a/crates/metrics/src/collectors/system_gpu.rs +++ b/crates/obs/src/metrics/collectors/system_gpu.rs @@ -22,7 +22,7 @@ //! # Example //! //! ```ignore -//! use rustfs_metrics::collectors::{GpuCollector, collect_gpu_metrics}; +//! use rustfs_obs::metrics::collectors::{GpuCollector, collect_gpu_metrics}; //! use sysinfo::Pid; //! //! let pid = sysinfo::get_current_pid().unwrap(); @@ -31,9 +31,9 @@ //! let metrics = collect_gpu_metrics(&stats, &labels); //! ``` -use crate::format::PrometheusMetric; +use crate::metrics::report::PrometheusMetric; -use crate::metrics_type::system_gpu::PROCESS_GPU_MEMORY_USAGE_MD; +use crate::metrics::schema::system_gpu::PROCESS_GPU_MEMORY_USAGE_MD; use nvml_wrapper::Nvml; @@ -98,7 +98,7 @@ impl GpuCollector { /// # Example /// /// ```ignore - /// use rustfs_metrics::collectors::GpuCollector; + /// use rustfs_obs::metrics::collectors::GpuCollector; /// use sysinfo::Pid; /// /// let pid = sysinfo::get_current_pid().unwrap(); @@ -163,7 +163,7 @@ impl GpuCollector { /// # Example /// /// ```ignore -/// use rustfs_metrics::collectors::{GpuStats, collect_gpu_metrics}; +/// use rustfs_obs::metrics::collectors::{GpuStats, collect_gpu_metrics}; /// /// let stats = GpuStats { memory_usage: 1024 }; /// let labels = vec![("process_pid", Cow::Borrowed("1234"))]; diff --git a/crates/metrics/src/collectors/system_memory.rs b/crates/obs/src/metrics/collectors/system_memory.rs similarity index 95% rename from crates/metrics/src/collectors/system_memory.rs rename to crates/obs/src/metrics/collectors/system_memory.rs index 54697cca5..78a4d625e 100644 --- a/crates/metrics/src/collectors/system_memory.rs +++ b/crates/obs/src/metrics/collectors/system_memory.rs @@ -22,9 +22,9 @@ //! This module provides both system-level and process-level memory metrics, //! with process-level metrics migrated from `rustfs-obs::system`. -use crate::format::PrometheusMetric; -use crate::metrics_type::system_memory::*; -use crate::metrics_type::system_process::{PROCESS_RESIDENT_MEMORY_BYTES_MD, PROCESS_VIRTUAL_MEMORY_BYTES_MD}; +use crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::system_memory::*; +use crate::metrics::schema::system_process::{PROCESS_RESIDENT_MEMORY_BYTES_MD, PROCESS_VIRTUAL_MEMORY_BYTES_MD}; use std::borrow::Cow; /// System memory statistics. @@ -103,7 +103,7 @@ pub fn collect_process_memory_metrics( #[cfg(test)] mod tests { use super::*; - use crate::format::report_metrics; + use crate::metrics::report::report_metrics; #[test] fn test_collect_memory_metrics() { diff --git a/crates/metrics/src/collectors/system_network.rs b/crates/obs/src/metrics/collectors/system_network.rs similarity index 96% rename from crates/metrics/src/collectors/system_network.rs rename to crates/obs/src/metrics/collectors/system_network.rs index bebac46c9..5112bfc21 100644 --- a/crates/metrics/src/collectors/system_network.rs +++ b/crates/obs/src/metrics/collectors/system_network.rs @@ -22,9 +22,9 @@ //! This module provides both system-level and process-level network metrics, //! with process-level metrics migrated from `rustfs-obs::system`. -use crate::format::PrometheusMetric; -use crate::metrics_type::system_network::*; -use crate::metrics_type::system_process::{PROCESS_NETWORK_IO_MD, PROCESS_NETWORK_IO_PER_INTERFACE_MD}; +use crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::system_network::*; +use crate::metrics::schema::system_process::{PROCESS_NETWORK_IO_MD, PROCESS_NETWORK_IO_PER_INTERFACE_MD}; use std::borrow::Cow; /// Network statistics for internode communication. @@ -126,7 +126,7 @@ pub fn collect_process_network_metrics( #[cfg(test)] mod tests { use super::*; - use crate::format::report_metrics; + use crate::metrics::report::report_metrics; #[test] fn test_collect_network_metrics() { diff --git a/crates/metrics/src/collectors/system_process.rs b/crates/obs/src/metrics/collectors/system_process.rs similarity index 98% rename from crates/metrics/src/collectors/system_process.rs rename to crates/obs/src/metrics/collectors/system_process.rs index dcb9624d2..ebdcb2516 100644 --- a/crates/metrics/src/collectors/system_process.rs +++ b/crates/obs/src/metrics/collectors/system_process.rs @@ -22,8 +22,8 @@ //! This module also provides process attribute collection for use as //! metric labels, migrated from `rustfs-obs::system`. -use crate::format::PrometheusMetric; -use crate::metrics_type::system_process::*; +use crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::system_process::*; use std::borrow::Cow; use sysinfo::{Pid, ProcessStatus, System}; @@ -230,7 +230,7 @@ pub fn collect_process_attributes() -> Result>>, value intern_string(cache, value) } -/// Report metrics using the `metrics` crate. -/// -/// This function iterates over the provided metrics and reports them using -/// the `metrics` crate's API. This allows integration with various metrics -/// exporters (e.g., Prometheus) that are configured globally. -/// pub fn report_metrics(metrics: &[PrometheusMetric]) { for metric in metrics { let name = into_static_str(&NAME_CACHE, &metric.name); let help = into_static_str(&HELP_CACHE, &metric.help); - // Register metric description (help text) - // Note: In a real-world scenario, descriptions should ideally be registered once at startup. - // However, the `metrics` crate handles duplicate registrations gracefully. match metric.metric_type { MetricType::Counter => describe_counter!(name, help), MetricType::Gauge => describe_gauge!(name, help), MetricType::Histogram => describe_histogram!(name, help), } - // Convert labels to the format expected by `metrics` crate let labels: Vec<(String, String)> = metric.labels.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect(); - // Report the metric value match metric.metric_type { MetricType::Counter => { - // Use counter! macro to get a handle, then set absolute value. - // Note: `metrics` crate counters are typically monotonic and support `increment`. - // Setting an absolute value directly requires `absolute` method if supported by the backend/handle, - // or we assume the value provided is the absolute count we want to report. - // - // Since `metrics` 0.21+, `Counter` has an `absolute` method which sets the counter to a specific value. - // This is useful for mirroring an external counter. let counter = counter!(name, &labels); counter.absolute(metric.value as u64); } @@ -91,30 +68,16 @@ pub fn report_metrics(metrics: &[PrometheusMetric]) { } } -/// A single Prometheus metric with labels and value. -/// -/// This struct is optimized for performance by using `Cow<'static, str>` for -/// the name and help text, which allows both static strings and owned strings. -/// Labels use `Cow<'static, str>` to avoid allocations when possible. #[derive(Debug, Clone)] pub struct PrometheusMetric { - /// The metric name (e.g., "http_requests_total"). pub name: Cow<'static, str>, - /// The type of this metric (counter, gauge, or histogram). pub metric_type: MetricType, - /// Human-readable description shown in Prometheus UI. pub help: Cow<'static, str>, - /// Key-value label pairs for this metric instance. - /// Uses Cow to avoid allocations for static label keys. pub labels: Vec<(&'static str, Cow<'static, str>)>, - /// The numeric value of this metric. pub value: f64, } impl PrometheusMetric { - /// Creates a new metric with the given name, type, help text, and value. - /// - /// Uses static strings to avoid heap allocations for metric metadata. #[inline] pub const fn new(name: &'static str, metric_type: MetricType, help: &'static str, value: f64) -> Self { Self { @@ -126,9 +89,6 @@ impl PrometheusMetric { } } - /// Creates a new metric with owned strings for name and help. - /// - /// Use this when the metric name or help text is dynamically generated. #[inline] pub fn new_owned(name: String, metric_type: MetricType, help: String, value: f64) -> Self { Self { @@ -140,12 +100,8 @@ impl PrometheusMetric { } } - /// Creates a new metric from a MetricDescriptor. - /// - /// This is the recommended way to create metrics when using MetricDescriptor - /// from the metrics_type module. #[inline] - pub fn from_descriptor(descriptor: &crate::MetricDescriptor, value: f64) -> Self { + pub fn from_descriptor(descriptor: &MetricDescriptor, value: f64) -> Self { let help = intern_string(&HELP_CACHE, &descriptor.help); Self { name: Cow::Owned(descriptor.get_full_metric_name()), @@ -156,7 +112,6 @@ impl PrometheusMetric { } } - /// Adds a single label with a static value to this metric. #[inline] #[allow(dead_code)] pub fn with_label(mut self, key: &'static str, value: impl Into>) -> Self { @@ -164,9 +119,6 @@ impl PrometheusMetric { self } - /// Adds a label with an owned string value. - /// - /// Use this when the label value is dynamically generated. #[inline] #[allow(dead_code)] pub fn with_label_owned(mut self, key: &'static str, value: String) -> Self { @@ -174,7 +126,6 @@ impl PrometheusMetric { self } - /// Sets all labels for this metric, replacing any existing labels. #[inline] #[allow(dead_code)] pub fn with_labels(mut self, labels: Vec<(&'static str, Cow<'static, str>)>) -> Self { @@ -186,7 +137,7 @@ impl PrometheusMetric { #[cfg(test)] mod tests { use super::*; - use crate::{MetricDescriptor, MetricName, MetricNamespace, MetricSubsystem}; + use crate::metrics::schema::{MetricName, MetricNamespace, MetricSubsystem}; #[test] fn from_descriptor_uses_prometheus_metric_names_for_all_types() { diff --git a/crates/metrics/src/collectors/global.rs b/crates/obs/src/metrics/scheduler.rs similarity index 73% rename from crates/metrics/src/collectors/global.rs rename to crates/obs/src/metrics/scheduler.rs index 0d31604bc..ff9acaed8 100644 --- a/crates/metrics/src/collectors/global.rs +++ b/crates/obs/src/metrics/scheduler.rs @@ -23,33 +23,43 @@ //! - Process disk I/O metrics //! - Process network I/O metrics -use crate::collectors::stats_collector::{ - collect_bucket_replication_bandwidth_stats, collect_bucket_stats, collect_cluster_stats, collect_disk_stats, - collect_process_stats, -}; -use crate::collectors::{ +use crate::metrics::collectors::{ + AuditTargetStats, + NotificationStats, + NotificationTargetStats, // System monitoring collectors (migrated from rustfs-obs::system) ProcessCpuStats, ProcessDiskStats, ProcessMemoryStats, ProcessNetworkStats, + collect_audit_metrics, collect_bucket_metrics, collect_bucket_replication_bandwidth_metrics, collect_cluster_metrics, collect_node_metrics, + collect_notification_metrics, + collect_notification_target_metrics, collect_process_cpu_metrics, collect_process_disk_metrics, collect_process_memory_metrics, + collect_process_metrics, collect_process_network_metrics, collect_resource_metrics, }; -use crate::constants::{ - DEFAULT_BUCKET_METRICS_INTERVAL, DEFAULT_BUCKET_REPLICATION_BANDWIDTH_METRICS_INTERVAL, DEFAULT_CLUSTER_METRICS_INTERVAL, - DEFAULT_NODE_METRICS_INTERVAL, DEFAULT_RESOURCE_METRICS_INTERVAL, ENV_BUCKET_METRICS_INTERVAL, +use crate::metrics::config::{ + DEFAULT_AUDIT_METRICS_INTERVAL, DEFAULT_BUCKET_METRICS_INTERVAL, DEFAULT_BUCKET_REPLICATION_BANDWIDTH_METRICS_INTERVAL, + DEFAULT_CLUSTER_METRICS_INTERVAL, DEFAULT_NODE_METRICS_INTERVAL, DEFAULT_NOTIFICATION_METRICS_INTERVAL, + DEFAULT_RESOURCE_METRICS_INTERVAL, ENV_AUDIT_METRICS_INTERVAL, ENV_BUCKET_METRICS_INTERVAL, ENV_BUCKET_REPLICATION_BANDWIDTH_METRICS_INTERVAL, ENV_CLUSTER_METRICS_INTERVAL, ENV_DEFAULT_METRICS_INTERVAL, - ENV_NODE_METRICS_INTERVAL, ENV_RESOURCE_METRICS_INTERVAL, + ENV_NODE_METRICS_INTERVAL, ENV_NOTIFICATION_METRICS_INTERVAL, ENV_RESOURCE_METRICS_INTERVAL, }; -use crate::format::report_metrics; +use crate::metrics::report::report_metrics; +use crate::metrics::stats_collector::{ + collect_bucket_replication_bandwidth_stats, collect_bucket_stats, collect_cluster_stats, collect_disk_stats, + collect_process_resource_and_system_stats, +}; +use rustfs_audit::audit_target_metrics; +use rustfs_notify::{notification_metrics_snapshot, notification_target_metrics}; use rustfs_utils::get_env_opt_u64; use std::borrow::Cow; use std::time::Duration; @@ -88,12 +98,14 @@ const LEGACY_SYSTEM_METRICS_INTERVAL: &str = "RUSTFS_OBS_METRICS_SYSTEM_INTERVAL /// - `RUSTFS_METRICS_NODE_INTERVAL` /// - `RUSTFS_METRICS_BUCKET_REPLICATION_BANDWIDTH_INTERVAL` /// - `RUSTFS_METRICS_RESOURCE_INTERVAL` -pub fn init_metrics_collectors(token: CancellationToken) { +pub fn init_metrics_runtime(token: CancellationToken) { const LEGACY_CLUSTER_INTERVAL: &str = "RUSTFS_METRICS_CLUSTER_INTERVAL"; const LEGACY_BUCKET_INTERVAL: &str = "RUSTFS_METRICS_BUCKET_INTERVAL"; const LEGACY_NODE_INTERVAL: &str = "RUSTFS_METRICS_NODE_INTERVAL"; const LEGACY_REPLICATION_BANDWIDTH_INTERVAL: &str = "RUSTFS_METRICS_BUCKET_REPLICATION_BANDWIDTH_INTERVAL"; const LEGACY_RESOURCE_INTERVAL: &str = "RUSTFS_METRICS_RESOURCE_INTERVAL"; + const LEGACY_AUDIT_INTERVAL: &str = "RUSTFS_METRICS_AUDIT_INTERVAL"; + const LEGACY_NOTIFICATION_INTERVAL: &str = "RUSTFS_METRICS_NOTIFICATION_INTERVAL"; const LEGACY_DEFAULT_INTERVAL: &str = "RUSTFS_METRICS_DEFAULT_INTERVAL"; /// Parse metrics interval from environment variables with fallback to default. @@ -126,6 +138,13 @@ pub fn init_metrics_collectors(token: CancellationToken) { let resource_interval = parse_metrics_interval(ENV_RESOURCE_METRICS_INTERVAL, LEGACY_RESOURCE_INTERVAL, DEFAULT_RESOURCE_METRICS_INTERVAL); + let audit_interval = + parse_metrics_interval(ENV_AUDIT_METRICS_INTERVAL, LEGACY_AUDIT_INTERVAL, DEFAULT_AUDIT_METRICS_INTERVAL); + let notification_interval = parse_metrics_interval( + ENV_NOTIFICATION_METRICS_INTERVAL, + LEGACY_NOTIFICATION_INTERVAL, + DEFAULT_NOTIFICATION_METRICS_INTERVAL, + ); // Spawn task for cluster metrics let token_clone = token.clone(); @@ -210,9 +229,9 @@ pub fn init_metrics_collectors(token: CancellationToken) { loop { tokio::select! { _ = interval.tick() => { - // Resource stats collection is synchronous but fast - let stats = collect_process_stats(); - let metrics = collect_resource_metrics(&stats); + let (resource_stats, process_stats) = collect_process_resource_and_system_stats(); + let mut metrics = collect_resource_metrics(&resource_stats); + metrics.extend(collect_process_metrics(&process_stats)); report_metrics(&metrics); } _ = token_clone.cancelled() => { @@ -223,6 +242,69 @@ pub fn init_metrics_collectors(token: CancellationToken) { } }); + // Spawn task for audit target delivery metrics + let token_clone = token.clone(); + tokio::spawn(async move { + let mut interval = tokio::time::interval(audit_interval); + loop { + tokio::select! { + _ = interval.tick() => { + let stats = audit_target_metrics().await + .into_iter() + .map(|snapshot| AuditTargetStats { + failed_messages: snapshot.failed_messages, + queue_length: snapshot.queue_length, + target_id: snapshot.target_id, + total_messages: snapshot.total_messages, + }) + .collect::>(); + let metrics = collect_audit_metrics(&stats); + report_metrics(&metrics); + } + _ = token_clone.cancelled() => { + warn!("Metrics collection for audit target stats cancelled."); + return; + } + } + } + }); + + // Spawn task for notification delivery metrics + let token_clone = token.clone(); + tokio::spawn(async move { + let mut interval = tokio::time::interval(notification_interval); + loop { + tokio::select! { + _ = interval.tick() => { + let snapshot = notification_metrics_snapshot(); + let mut metrics = collect_notification_metrics(&NotificationStats { + current_send_in_progress: snapshot.current_send_in_progress, + events_errors_total: snapshot.events_errors_total, + events_sent_total: snapshot.events_sent_total, + events_skipped_total: snapshot.events_skipped_total, + }); + + let target_stats = notification_target_metrics().await + .into_iter() + .map(|snapshot| NotificationTargetStats { + failed_messages: snapshot.failed_messages, + queue_length: snapshot.queue_length, + target_id: snapshot.target_id, + target_type: snapshot.target_type, + total_messages: snapshot.total_messages, + }) + .collect::>(); + metrics.extend(collect_notification_target_metrics(&target_stats)); + report_metrics(&metrics); + } + _ = token_clone.cancelled() => { + warn!("Metrics collection for notification stats cancelled."); + return; + } + } + } + }); + // Spawn task for system monitoring metrics (migrated from rustfs-obs::system) let system_interval = get_env_opt_u64(ENV_SYSTEM_METRICS_INTERVAL) .or_else(|| get_env_opt_u64(LEGACY_SYSTEM_METRICS_INTERVAL).map(|ms| ms / 1000)) // Convert ms to seconds @@ -259,6 +341,11 @@ pub fn init_metrics_collectors(token: CancellationToken) { }); } +/// Backward-compatible alias kept during migration. +pub fn init_metrics_collectors(token: CancellationToken) { + init_metrics_runtime(token); +} + /// Collect all system monitoring metrics for a process. /// /// This function collects CPU, memory, disk I/O, and network I/O metrics @@ -269,7 +356,7 @@ pub fn init_metrics_collectors(token: CancellationToken) { /// /// # Returns /// A vector of Prometheus metrics for the process. -fn collect_system_monitoring_metrics(pid: Pid) -> Vec { +fn collect_system_monitoring_metrics(pid: Pid) -> Vec { let mut metrics = Vec::new(); let mut system = System::new(); @@ -331,7 +418,7 @@ fn collect_system_monitoring_metrics(pid: Pid) -> Vec match collector.collect() { diff --git a/crates/metrics/src/metrics_type/audit.rs b/crates/obs/src/metrics/schema/audit.rs similarity index 100% rename from crates/metrics/src/metrics_type/audit.rs rename to crates/obs/src/metrics/schema/audit.rs diff --git a/crates/metrics/src/metrics_type/bucket.rs b/crates/obs/src/metrics/schema/bucket.rs similarity index 100% rename from crates/metrics/src/metrics_type/bucket.rs rename to crates/obs/src/metrics/schema/bucket.rs diff --git a/crates/metrics/src/metrics_type/bucket_replication.rs b/crates/obs/src/metrics/schema/bucket_replication.rs similarity index 100% rename from crates/metrics/src/metrics_type/bucket_replication.rs rename to crates/obs/src/metrics/schema/bucket_replication.rs diff --git a/crates/metrics/src/metrics_type/cluster.rs b/crates/obs/src/metrics/schema/cluster.rs similarity index 100% rename from crates/metrics/src/metrics_type/cluster.rs rename to crates/obs/src/metrics/schema/cluster.rs diff --git a/crates/metrics/src/metrics_type/cluster_config.rs b/crates/obs/src/metrics/schema/cluster_config.rs similarity index 100% rename from crates/metrics/src/metrics_type/cluster_config.rs rename to crates/obs/src/metrics/schema/cluster_config.rs diff --git a/crates/metrics/src/metrics_type/cluster_erasure_set.rs b/crates/obs/src/metrics/schema/cluster_erasure_set.rs similarity index 100% rename from crates/metrics/src/metrics_type/cluster_erasure_set.rs rename to crates/obs/src/metrics/schema/cluster_erasure_set.rs diff --git a/crates/metrics/src/metrics_type/cluster_health.rs b/crates/obs/src/metrics/schema/cluster_health.rs similarity index 100% rename from crates/metrics/src/metrics_type/cluster_health.rs rename to crates/obs/src/metrics/schema/cluster_health.rs diff --git a/crates/metrics/src/metrics_type/cluster_iam.rs b/crates/obs/src/metrics/schema/cluster_iam.rs similarity index 100% rename from crates/metrics/src/metrics_type/cluster_iam.rs rename to crates/obs/src/metrics/schema/cluster_iam.rs diff --git a/crates/metrics/src/metrics_type/cluster_notification.rs b/crates/obs/src/metrics/schema/cluster_notification.rs similarity index 89% rename from crates/metrics/src/metrics_type/cluster_notification.rs rename to crates/obs/src/metrics/schema/cluster_notification.rs index 58797f9a6..5fbedc67c 100644 --- a/crates/metrics/src/metrics_type/cluster_notification.rs +++ b/crates/obs/src/metrics/schema/cluster_notification.rs @@ -14,11 +14,11 @@ #![allow(dead_code)] -use crate::{MetricDescriptor, MetricName, new_counter_md, subsystems}; +use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems}; use std::sync::LazyLock; pub static NOTIFICATION_CURRENT_SEND_IN_PROGRESS_MD: LazyLock = LazyLock::new(|| { - new_counter_md( + new_gauge_md( MetricName::NotificationCurrentSendInProgress, "Number of concurrent async Send calls active to all targets", &[], @@ -47,7 +47,7 @@ pub static NOTIFICATION_EVENTS_SENT_TOTAL_MD: LazyLock = LazyL pub static NOTIFICATION_EVENTS_SKIPPED_TOTAL_MD: LazyLock = LazyLock::new(|| { new_counter_md( MetricName::NotificationEventsSkippedTotal, - "Events that were skipped to be sent to the targets due to the in-memory queue being full", + "Notification dispatch attempts skipped before delivery", &[], subsystems::NOTIFICATION, ) diff --git a/crates/metrics/src/metrics_type/cluster_usage.rs b/crates/obs/src/metrics/schema/cluster_usage.rs similarity index 100% rename from crates/metrics/src/metrics_type/cluster_usage.rs rename to crates/obs/src/metrics/schema/cluster_usage.rs diff --git a/crates/metrics/src/metrics_type/entry/descriptor.rs b/crates/obs/src/metrics/schema/entry/descriptor.rs similarity index 100% rename from crates/metrics/src/metrics_type/entry/descriptor.rs rename to crates/obs/src/metrics/schema/entry/descriptor.rs diff --git a/crates/metrics/src/metrics_type/entry/metric_name.rs b/crates/obs/src/metrics/schema/entry/metric_name.rs similarity index 98% rename from crates/metrics/src/metrics_type/entry/metric_name.rs rename to crates/obs/src/metrics/schema/entry/metric_name.rs index 7f1eeb9bc..36b2185d8 100644 --- a/crates/metrics/src/metrics_type/entry/metric_name.rs +++ b/crates/obs/src/metrics/schema/entry/metric_name.rs @@ -220,6 +220,9 @@ pub enum MetricName { NotificationEventsErrorsTotal, NotificationEventsSentTotal, NotificationEventsSkippedTotal, + NotificationTargetFailedMessages, + NotificationTargetQueueLength, + NotificationTargetTotalMessages, // Metrics related to the usage of cluster objects UsageSinceLastUpdateSeconds, @@ -247,11 +250,6 @@ pub enum MetricName { IlmTransitionMissedImmediateTasks, IlmVersionsScanned, - // Webhook logs - WebhookQueueLength, - WebhookTotalMessages, - WebhookFailedMessages, - // Copy the relevant metrics ReplicationAverageActiveWorkers, ReplicationAverageQueuedBytes, @@ -558,6 +556,9 @@ impl MetricName { Self::NotificationEventsErrorsTotal => "events_errors_total".to_string(), Self::NotificationEventsSentTotal => "events_sent_total".to_string(), Self::NotificationEventsSkippedTotal => "events_skipped_total".to_string(), + Self::NotificationTargetFailedMessages => "failed_messages".to_string(), + Self::NotificationTargetQueueLength => "target_queue_length".to_string(), + Self::NotificationTargetTotalMessages => "total_messages".to_string(), // Metrics related to the usage of cluster objects Self::UsageSinceLastUpdateSeconds => "since_last_update_seconds".to_string(), @@ -585,11 +586,6 @@ impl MetricName { Self::IlmTransitionMissedImmediateTasks => "transition_missed_immediate_tasks".to_string(), Self::IlmVersionsScanned => "versions_scanned".to_string(), - // Webhook logs - Self::WebhookQueueLength => "queue_length".to_string(), - Self::WebhookTotalMessages => "total_messages".to_string(), - Self::WebhookFailedMessages => "failed_messages".to_string(), - // Copy the relevant metrics Self::ReplicationAverageActiveWorkers => "average_active_workers".to_string(), Self::ReplicationAverageQueuedBytes => "average_queued_bytes".to_string(), diff --git a/crates/metrics/src/metrics_type/entry/metric_type.rs b/crates/obs/src/metrics/schema/entry/metric_type.rs similarity index 100% rename from crates/metrics/src/metrics_type/entry/metric_type.rs rename to crates/obs/src/metrics/schema/entry/metric_type.rs diff --git a/crates/metrics/src/metrics_type/entry/mod.rs b/crates/obs/src/metrics/schema/entry/mod.rs similarity index 100% rename from crates/metrics/src/metrics_type/entry/mod.rs rename to crates/obs/src/metrics/schema/entry/mod.rs diff --git a/crates/metrics/src/metrics_type/entry/namespace.rs b/crates/obs/src/metrics/schema/entry/namespace.rs similarity index 100% rename from crates/metrics/src/metrics_type/entry/namespace.rs rename to crates/obs/src/metrics/schema/entry/namespace.rs diff --git a/crates/metrics/src/metrics_type/entry/path_utils.rs b/crates/obs/src/metrics/schema/entry/path_utils.rs similarity index 100% rename from crates/metrics/src/metrics_type/entry/path_utils.rs rename to crates/obs/src/metrics/schema/entry/path_utils.rs diff --git a/crates/metrics/src/metrics_type/entry/subsystem.rs b/crates/obs/src/metrics/schema/entry/subsystem.rs similarity index 97% rename from crates/metrics/src/metrics_type/entry/subsystem.rs rename to crates/obs/src/metrics/schema/entry/subsystem.rs index 2567560b9..2a1d33063 100644 --- a/crates/metrics/src/metrics_type/entry/subsystem.rs +++ b/crates/obs/src/metrics/schema/entry/subsystem.rs @@ -47,7 +47,6 @@ pub enum MetricSubsystem { // other service related subsystems Ilm, Audit, - LoggerWebhook, Replication, Notification, Scanner, @@ -89,7 +88,6 @@ impl MetricSubsystem { // other service related subsystems Self::Ilm => "/ilm", Self::Audit => "/audit", - Self::LoggerWebhook => "/logger/webhook", Self::Replication => "/replication", Self::Notification => "/notification", Self::Scanner => "/scanner", @@ -137,7 +135,6 @@ impl MetricSubsystem { // Other service-related subsystems "/ilm" => Self::Ilm, "/audit" => Self::Audit, - "/logger/webhook" => Self::LoggerWebhook, "/replication" => Self::Replication, "/notification" => Self::Notification, "/scanner" => Self::Scanner, @@ -199,7 +196,6 @@ pub mod subsystems { pub const CLUSTER_CONFIG: MetricSubsystem = MetricSubsystem::ClusterConfig; pub const ILM: MetricSubsystem = MetricSubsystem::Ilm; pub const AUDIT: MetricSubsystem = MetricSubsystem::Audit; - pub const LOGGER_WEBHOOK: MetricSubsystem = MetricSubsystem::LoggerWebhook; pub const REPLICATION: MetricSubsystem = MetricSubsystem::Replication; pub const NOTIFICATION: MetricSubsystem = MetricSubsystem::Notification; pub const SCANNER: MetricSubsystem = MetricSubsystem::Scanner; diff --git a/crates/metrics/src/metrics_type/ilm.rs b/crates/obs/src/metrics/schema/ilm.rs similarity index 100% rename from crates/metrics/src/metrics_type/ilm.rs rename to crates/obs/src/metrics/schema/ilm.rs diff --git a/crates/metrics/src/metrics_type/mod.rs b/crates/obs/src/metrics/schema/mod.rs similarity index 97% rename from crates/metrics/src/metrics_type/mod.rs rename to crates/obs/src/metrics/schema/mod.rs index 603a7b83e..c47f990e1 100644 --- a/crates/metrics/src/metrics_type/mod.rs +++ b/crates/obs/src/metrics/schema/mod.rs @@ -24,9 +24,9 @@ pub mod cluster_notification; pub mod cluster_usage; pub mod entry; pub mod ilm; -pub mod logger_webhook; pub mod node_bucket; pub mod node_disk; +pub mod notification_target; pub mod process_resource; pub mod replication; pub mod request; diff --git a/crates/metrics/src/metrics_type/node_bucket.rs b/crates/obs/src/metrics/schema/node_bucket.rs similarity index 100% rename from crates/metrics/src/metrics_type/node_bucket.rs rename to crates/obs/src/metrics/schema/node_bucket.rs diff --git a/crates/metrics/src/metrics_type/node_disk.rs b/crates/obs/src/metrics/schema/node_disk.rs similarity index 100% rename from crates/metrics/src/metrics_type/node_disk.rs rename to crates/obs/src/metrics/schema/node_disk.rs diff --git a/crates/obs/src/metrics/schema/notification_target.rs b/crates/obs/src/metrics/schema/notification_target.rs new file mode 100644 index 000000000..f3af7d89f --- /dev/null +++ b/crates/obs/src/metrics/schema/notification_target.rs @@ -0,0 +1,50 @@ +// 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. + +#![allow(dead_code)] + +use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems}; +use std::sync::LazyLock; + +pub const TARGET_ID: &str = "target_id"; +pub const TARGET_TYPE: &str = "target_type"; + +const NOTIFICATION_TARGET_LABELS: [&str; 2] = [TARGET_ID, TARGET_TYPE]; + +pub static NOTIFICATION_TARGET_FAILED_MESSAGES_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::NotificationTargetFailedMessages, + "Total number of notification messages that permanently failed to send", + &NOTIFICATION_TARGET_LABELS, + subsystems::NOTIFICATION, + ) +}); + +pub static NOTIFICATION_TARGET_QUEUE_LENGTH_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::NotificationTargetQueueLength, + "Number of queued notification messages pending delivery", + &NOTIFICATION_TARGET_LABELS, + subsystems::NOTIFICATION, + ) +}); + +pub static NOTIFICATION_TARGET_TOTAL_MESSAGES_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::NotificationTargetTotalMessages, + "Total number of notification messages successfully delivered", + &NOTIFICATION_TARGET_LABELS, + subsystems::NOTIFICATION, + ) +}); diff --git a/crates/metrics/src/metrics_type/process_resource.rs b/crates/obs/src/metrics/schema/process_resource.rs similarity index 100% rename from crates/metrics/src/metrics_type/process_resource.rs rename to crates/obs/src/metrics/schema/process_resource.rs diff --git a/crates/metrics/src/metrics_type/replication.rs b/crates/obs/src/metrics/schema/replication.rs similarity index 100% rename from crates/metrics/src/metrics_type/replication.rs rename to crates/obs/src/metrics/schema/replication.rs diff --git a/crates/metrics/src/metrics_type/request.rs b/crates/obs/src/metrics/schema/request.rs similarity index 100% rename from crates/metrics/src/metrics_type/request.rs rename to crates/obs/src/metrics/schema/request.rs diff --git a/crates/metrics/src/metrics_type/scanner.rs b/crates/obs/src/metrics/schema/scanner.rs similarity index 100% rename from crates/metrics/src/metrics_type/scanner.rs rename to crates/obs/src/metrics/schema/scanner.rs diff --git a/crates/metrics/src/metrics_type/system_cpu.rs b/crates/obs/src/metrics/schema/system_cpu.rs similarity index 100% rename from crates/metrics/src/metrics_type/system_cpu.rs rename to crates/obs/src/metrics/schema/system_cpu.rs diff --git a/crates/metrics/src/metrics_type/system_drive.rs b/crates/obs/src/metrics/schema/system_drive.rs similarity index 100% rename from crates/metrics/src/metrics_type/system_drive.rs rename to crates/obs/src/metrics/schema/system_drive.rs diff --git a/crates/metrics/src/metrics_type/system_gpu.rs b/crates/obs/src/metrics/schema/system_gpu.rs similarity index 100% rename from crates/metrics/src/metrics_type/system_gpu.rs rename to crates/obs/src/metrics/schema/system_gpu.rs diff --git a/crates/metrics/src/metrics_type/system_memory.rs b/crates/obs/src/metrics/schema/system_memory.rs similarity index 100% rename from crates/metrics/src/metrics_type/system_memory.rs rename to crates/obs/src/metrics/schema/system_memory.rs diff --git a/crates/metrics/src/metrics_type/system_network.rs b/crates/obs/src/metrics/schema/system_network.rs similarity index 100% rename from crates/metrics/src/metrics_type/system_network.rs rename to crates/obs/src/metrics/schema/system_network.rs diff --git a/crates/metrics/src/metrics_type/system_process.rs b/crates/obs/src/metrics/schema/system_process.rs similarity index 100% rename from crates/metrics/src/metrics_type/system_process.rs rename to crates/obs/src/metrics/schema/system_process.rs diff --git a/crates/metrics/src/collectors/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs similarity index 71% rename from crates/metrics/src/collectors/stats_collector.rs rename to crates/obs/src/metrics/stats_collector.rs index 03e6b0951..3d86c694d 100644 --- a/crates/metrics/src/collectors/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -20,27 +20,18 @@ //! RustFS internal sources (storage layer, bucket monitor, system info) //! and convert them to the Stats structs used by collectors. -use crate::collectors::{BucketReplicationBandwidthStats, BucketStats, ClusterStats, DiskStats, ResourceStats}; +use crate::metrics::collectors::{ + BucketReplicationBandwidthStats, BucketStats, ClusterStats, DiskStats, ProcessStats, ProcessStatusType, ResourceStats, +}; use rustfs_ecstore::bucket::metadata_sys::get_quota_config; use rustfs_ecstore::data_usage::load_data_usage_from_backend; use rustfs_ecstore::global::get_global_bucket_monitor; use rustfs_ecstore::pools::{get_total_usable_capacity, get_total_usable_capacity_free}; use rustfs_ecstore::store_api::{BucketOperations, BucketOptions}; use rustfs_ecstore::{StorageAPI, new_object_layer_fn}; -use std::sync::OnceLock; -use std::time::Instant; -use sysinfo::{Pid, ProcessRefreshKind, ProcessesToUpdate, System}; +use rustfs_io_metrics::{ProcessStatusSnapshot, snapshot_process_resource_and_system}; use tracing::{instrument, warn}; -/// Process start time for calculating uptime. -static PROCESS_START: OnceLock = OnceLock::new(); - -/// Get the process start time, initializing it on first call. -#[inline] -fn get_process_start() -> &'static Instant { - PROCESS_START.get_or_init(Instant::now) -} - /// Collect cluster statistics from the storage layer. #[instrument] pub async fn collect_cluster_stats() -> ClusterStats { @@ -197,37 +188,55 @@ pub async fn collect_disk_stats() -> Vec { .collect() } +/// Collect resource and process statistics for the current process in one sysinfo refresh. +#[inline] +pub fn collect_process_resource_and_system_stats() -> (ResourceStats, ProcessStats) { + let (resource_snapshot, process_snapshot) = snapshot_process_resource_and_system(); + let status = match process_snapshot.status { + ProcessStatusSnapshot::Running => ProcessStatusType::Running, + ProcessStatusSnapshot::Sleeping => ProcessStatusType::Sleeping, + ProcessStatusSnapshot::Zombie => ProcessStatusType::Zombie, + ProcessStatusSnapshot::Other => ProcessStatusType::Other, + }; + + let resource_stats = ResourceStats { + cpu_percent: resource_snapshot.cpu_percent, + memory_bytes: resource_snapshot.memory_bytes, + uptime_seconds: resource_snapshot.uptime_seconds, + }; + let process_stats = ProcessStats { + locks_read_total: process_snapshot.locks_read_total, + locks_write_total: process_snapshot.locks_write_total, + cpu_total_seconds: process_snapshot.cpu_total_seconds, + file_descriptor_limit_total: process_snapshot.file_descriptor_limit_total, + file_descriptor_open_total: process_snapshot.file_descriptor_open_total, + go_routine_total: process_snapshot.go_routine_total, + io_rchar_bytes: process_snapshot.io_rchar_bytes, + io_read_bytes: process_snapshot.io_read_bytes, + io_wchar_bytes: process_snapshot.io_wchar_bytes, + io_write_bytes: process_snapshot.io_write_bytes, + resident_memory_bytes: process_snapshot.resident_memory_bytes, + start_time_seconds: process_snapshot.start_time_seconds, + status, + status_value: process_snapshot.status_value, + syscall_read_total: process_snapshot.syscall_read_total, + syscall_write_total: process_snapshot.syscall_write_total, + uptime_seconds: process_snapshot.uptime_seconds, + virtual_memory_bytes: process_snapshot.virtual_memory_bytes, + virtual_memory_max_bytes: process_snapshot.virtual_memory_max_bytes, + }; + + (resource_stats, process_stats) +} + /// Collect resource statistics for the current process. -/// -/// Collects: -/// - Uptime: Calculated from process start time -/// - Memory: Process resident set size from sysinfo -/// - CPU: Process CPU usage percentage from sysinfo #[inline] pub fn collect_process_stats() -> ResourceStats { - let uptime_seconds = get_process_start().elapsed().as_secs(); - - // Use sysinfo for process metrics - let mut sys = System::new(); - let pid = Pid::from_u32(std::process::id()); - sys.refresh_processes_specifics( - ProcessesToUpdate::Some(&[pid]), - true, - ProcessRefreshKind::nothing().with_cpu().with_memory(), - ); - - if let Some(process) = sys.process(pid) { - ResourceStats { - cpu_percent: process.cpu_usage() as f64, - memory_bytes: process.memory(), - uptime_seconds, - } - } else { - // Fallback if process info unavailable - ResourceStats { - cpu_percent: 0.0, - memory_bytes: 0, - uptime_seconds, - } - } + collect_process_resource_and_system_stats().0 +} + +/// Collect process statistics for the current process. +#[inline] +pub fn collect_process_system_stats() -> ProcessStats { + collect_process_resource_and_system_stats().1 } diff --git a/crates/targets/src/error.rs b/crates/targets/src/error.rs index 2e6d60916..c054cd3b2 100644 --- a/crates/targets/src/error.rs +++ b/crates/targets/src/error.rs @@ -82,6 +82,9 @@ pub enum TargetError { #[error("Target is disabled")] Disabled, + #[error("Queued payload dropped: {0}")] + Dropped(String), + #[error("Configuration parsing error: {0}")] ParseError(String), diff --git a/crates/targets/src/lib.rs b/crates/targets/src/lib.rs index 015159caa..ef684e23d 100644 --- a/crates/targets/src/lib.rs +++ b/crates/targets/src/lib.rs @@ -24,7 +24,7 @@ pub use error::{StoreError, TargetError}; pub use rustfs_s3_common::EventName; use serde::{Deserialize, Serialize}; pub use sys::user_agent::*; -pub use target::Target; +pub use target::{Target, TargetDeliverySnapshot}; /// Represents a log of events for sending to targets #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/crates/targets/src/target/mod.rs b/crates/targets/src/target/mod.rs index dc16dcae2..8412b82a2 100644 --- a/crates/targets/src/target/mod.rs +++ b/crates/targets/src/target/mod.rs @@ -21,12 +21,49 @@ use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; use std::fmt::Formatter; use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{SystemTime, UNIX_EPOCH}; use tracing::warn; pub mod mqtt; pub mod webhook; +/// A read-only snapshot of delivery counters for a target. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct TargetDeliverySnapshot { + pub failed_messages: u64, + pub queue_length: u64, + pub total_messages: u64, +} + +/// Shared target delivery counters. +#[derive(Debug, Default)] +pub struct TargetDeliveryCounters { + failed_messages: AtomicU64, + total_messages: AtomicU64, +} + +impl TargetDeliveryCounters { + #[inline] + pub fn record_success(&self) { + self.total_messages.fetch_add(1, Ordering::Relaxed); + } + + #[inline] + pub fn record_final_failure(&self) { + self.failed_messages.fetch_add(1, Ordering::Relaxed); + } + + #[inline] + pub fn snapshot(&self, queue_length: u64) -> TargetDeliverySnapshot { + TargetDeliverySnapshot { + failed_messages: self.failed_messages.load(Ordering::Relaxed), + queue_length, + total_messages: self.total_messages.load(Ordering::Relaxed), + } + } +} + /// Trait for notification targets #[async_trait] pub trait Target: Send + Sync + 'static @@ -70,8 +107,9 @@ where "Failed to delete invalid queued payload {key} after decode error '{err}': {delete_err}" )) })?; + self.record_final_failure(); warn!("Dropped invalid queued payload {key}: {err}"); - return Ok(()); + return Err(TargetError::Dropped(format!("Dropped invalid queued payload {key}: {err}"))); } }; @@ -96,6 +134,17 @@ where /// Check if the target is enabled fn is_enabled(&self) -> bool; + + /// Returns a read-only delivery snapshot for metrics collection. + fn delivery_snapshot(&self) -> TargetDeliverySnapshot { + TargetDeliverySnapshot { + queue_length: self.store().map_or(0, |store| store.len() as u64), + ..TargetDeliverySnapshot::default() + } + } + + /// Records a final, non-retryable delivery failure for metrics collection. + fn record_final_failure(&self) {} } #[derive(Debug, Serialize, Clone, Deserialize)] diff --git a/crates/targets/src/target/mqtt.rs b/crates/targets/src/target/mqtt.rs index beb37f395..2c2ce7759 100644 --- a/crates/targets/src/target/mqtt.rs +++ b/crates/targets/src/target/mqtt.rs @@ -17,7 +17,10 @@ use crate::{ arn::TargetID, error::TargetError, store::{Key, QueueStore, Store}, - target::{ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetType}, + target::{ + ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot, + TargetType, + }, }; use async_trait::async_trait; use hyper_rustls::ConfigBuilderExt; @@ -494,6 +497,7 @@ where store: Option + Send + Sync>>, connected: Arc, bg_task_manager: Arc, + delivery_counters: Arc, _phantom: PhantomData, } @@ -546,6 +550,7 @@ where store: queue_store, connected: Arc::new(AtomicBool::new(false)), bg_task_manager, + delivery_counters: Arc::new(TargetDeliveryCounters::default()), _phantom: PhantomData, }) } @@ -685,6 +690,7 @@ where })?; debug!(target_id = %self.id, topic = %self.args.topic, "Event published to MQTT topic"); + self.delivery_counters.record_success(); Ok(()) } @@ -696,6 +702,7 @@ where store: self.store.as_ref().map(|s| s.boxed_clone()), connected: self.connected.clone(), bg_task_manager: self.bg_task_manager.clone(), + delivery_counters: self.delivery_counters.clone(), _phantom: PhantomData, }) } @@ -890,21 +897,31 @@ where #[instrument(skip(self, event), fields(target_id = %self.id))] async fn save(&self, event: Arc>) -> Result<(), TargetError> { - let queued = self.build_queued_payload(&event)?; + let queued = match self.build_queued_payload(&event) { + Ok(queued) => queued, + Err(err) => { + self.delivery_counters.record_final_failure(); + return Err(err); + } + }; if let Some(store) = &self.store { debug!(target_id = %self.id, "Event saved to store start"); - match store.put_raw( - &queued - .encode() - .map_err(|e| TargetError::Storage(format!("Failed to encode queued payload: {e}")))?, - ) { + let encoded = match queued.encode() { + Ok(encoded) => encoded, + Err(err) => { + self.delivery_counters.record_final_failure(); + return Err(TargetError::Storage(format!("Failed to encode queued payload: {err}"))); + } + }; + match store.put_raw(&encoded) { Ok(_) => { debug!(target_id = %self.id, "Event saved to store for MQTT target successfully."); Ok(()) } Err(e) => { error!(target_id = %self.id, error = %e, "Failed to save event to store"); + self.delivery_counters.record_final_failure(); return Err(TargetError::Storage(format!("Failed to save event to store: {e}"))); } } @@ -920,15 +937,21 @@ where Ok(_) => debug!(target_id = %self.id, "MQTT target initialized successfully."), Err(e) => { error!(target_id = %self.id, error = %e, "Failed to initialize MQTT target."); + self.delivery_counters.record_final_failure(); return Err(TargetError::NotConnected); } } if !self.connected.load(Ordering::SeqCst) { error!(target_id = %self.id, "Cannot save (send directly) as target is not active after init attempt."); + self.delivery_counters.record_final_failure(); return Err(TargetError::NotConnected); } } - self.send_body(queued.body, &queued.meta).await + if let Err(err) = self.send_body(queued.body, &queued.meta).await { + self.delivery_counters.record_final_failure(); + return Err(err); + } + Ok(()) } } @@ -1017,6 +1040,15 @@ where fn is_enabled(&self) -> bool { self.args.enable } + + fn delivery_snapshot(&self) -> TargetDeliverySnapshot { + self.delivery_counters + .snapshot(self.store.as_deref().map_or(0, |store| store.len() as u64)) + } + + fn record_final_failure(&self) { + self.delivery_counters.record_final_failure(); + } } #[cfg(test)] diff --git a/crates/targets/src/target/webhook.rs b/crates/targets/src/target/webhook.rs index 875c6499c..79b6ba520 100644 --- a/crates/targets/src/target/webhook.rs +++ b/crates/targets/src/target/webhook.rs @@ -17,7 +17,10 @@ use crate::{ arn::TargetID, error::TargetError, store::{Key, QueueStore, Store}, - target::{ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetType}, + target::{ + ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot, + TargetType, + }, }; use async_trait::async_trait; use reqwest::{Client, StatusCode, Url}; @@ -108,6 +111,7 @@ where store: Option + Send + Sync>>, initialized: AtomicBool, cancel_sender: mpsc::Sender<()>, + delivery_counters: Arc, _phantom: PhantomData, } @@ -124,6 +128,7 @@ where store: self.store.as_ref().map(|s| s.boxed_clone()), initialized: AtomicBool::new(self.initialized.load(Ordering::SeqCst)), cancel_sender: self.cancel_sender.clone(), + delivery_counters: Arc::clone(&self.delivery_counters), _phantom: PhantomData, }) } @@ -172,6 +177,7 @@ where store: queue_store, initialized: AtomicBool::new(false), cancel_sender, + delivery_counters: Arc::new(TargetDeliveryCounters::default()), _phantom: PhantomData, }) } @@ -320,6 +326,7 @@ where let status = resp.status(); if status.is_success() { debug!("Event sent to webhook target: {}", self.id); + self.delivery_counters.record_success(); Ok(()) } else if status == StatusCode::FORBIDDEN { Err(TargetError::Authentication(format!( @@ -369,16 +376,26 @@ where } async fn save(&self, event: Arc>) -> Result<(), TargetError> { - let queued = self.build_queued_payload(&event)?; + let queued = match self.build_queued_payload(&event) { + Ok(queued) => queued, + Err(err) => { + self.delivery_counters.record_final_failure(); + return Err(err); + } + }; if let Some(store) = &self.store { - store - .put_raw( - &queued - .encode() - .map_err(|e| TargetError::Storage(format!("Failed to encode queued payload: {e}")))?, - ) - .map_err(|e| TargetError::Storage(format!("Failed to save event to store: {e}")))?; + let encoded = match queued.encode() { + Ok(encoded) => encoded, + Err(err) => { + self.delivery_counters.record_final_failure(); + return Err(TargetError::Storage(format!("Failed to encode queued payload: {err}"))); + } + }; + if let Err(e) = store.put_raw(&encoded) { + self.delivery_counters.record_final_failure(); + return Err(TargetError::Storage(format!("Failed to save event to store: {e}"))); + } debug!("Event saved to store for target: {}", self.id); Ok(()) } else { @@ -386,10 +403,15 @@ where Ok(_) => (), Err(e) => { error!("Failed to initialize Webhook target {}: {}", self.id.id, e); + self.delivery_counters.record_final_failure(); return Err(TargetError::NotConnected); } } - self.send_body(queued.body, &queued.meta).await + if let Err(err) = self.send_body(queued.body, &queued.meta).await { + self.delivery_counters.record_final_failure(); + return Err(err); + } + Ok(()) } } @@ -443,6 +465,15 @@ where fn is_enabled(&self) -> bool { self.args.enable } + + fn delivery_snapshot(&self) -> TargetDeliverySnapshot { + self.delivery_counters + .snapshot(self.store.as_deref().map_or(0, |store| store.len() as u64)) + } + + fn record_final_failure(&self) { + self.delivery_counters.record_final_failure(); + } } #[cfg(test)] diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index e0111f024..16b008ee2 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -43,16 +43,15 @@ bench = false required-features = ["manual-test-runners"] [features] -default = ["direct-io"] -metrics-gpu = ["rustfs-metrics/gpu"] +default = [] +metrics-gpu = ["rustfs-obs/gpu"] ftps = ["rustfs-protocols/ftps"] swift = ["rustfs-protocols/swift"] webdav = ["rustfs-protocols/webdav"] license = [] -direct-io = [] # Aligned direct I/O reader support (uses aligned pread, does not set O_DIRECT) io-scheduler-debug = [] # Enable debug information in I/O scheduler tracing-chunk-debug = [] # Enable per-chunk tracing in data plane (high noise, for debugging only) -full = ["metrics-gpu", "ftps", "swift", "webdav", "direct-io"] +full = ["metrics-gpu", "ftps", "swift", "webdav"] manual-test-runners = [] [lints] @@ -74,7 +73,6 @@ rustfs-keystone = { workspace = true } rustfs-kms = { workspace = true } rustfs-lock.workspace = true rustfs-madmin = { workspace = true } -rustfs-metrics = { workspace = true } rustfs-notify = { workspace = true } rustfs-obs = { workspace = true } rustfs-policy = { workspace = true } diff --git a/rustfs/src/config/info.rs b/rustfs/src/config/info.rs index 92315ad19..4ff0e89d9 100644 --- a/rustfs/src/config/info.rs +++ b/rustfs/src/config/info.rs @@ -590,20 +590,13 @@ struct FeatureSpec { default_enabled: bool, } -fn feature_specs() -> [FeatureSpec; 9] { +fn feature_specs() -> [FeatureSpec; 8] { [ - FeatureSpec { - name: "direct-io", - enabled: cfg!(feature = "direct-io"), - description: "Aligned pread-based direct I/O reader support", - dependencies: "(none)", - default_enabled: true, - }, FeatureSpec { name: "metrics-gpu", enabled: cfg!(feature = "metrics-gpu"), description: "Metrics GPU support", - dependencies: "rustfs-metrics/gpu", + dependencies: "rustfs-obs/gpu", default_enabled: false, }, FeatureSpec { @@ -652,7 +645,7 @@ fn feature_specs() -> [FeatureSpec; 9] { name: "full", enabled: cfg!(feature = "full"), description: "All features enabled", - dependencies: "metrics-gpu + ftps + swift + webdav + direct-io", + dependencies: "metrics-gpu + ftps + swift + webdav", default_enabled: false, }, ] @@ -926,13 +919,13 @@ mod tests { let info = collect_deps_info_json(); let feature_names: Vec<_> = info.features.iter().map(|feature| feature.name).collect(); - assert_eq!(info.total_count, 9); - assert_eq!(info.features.len(), 9); - assert!(feature_names.contains(&"direct-io")); + assert_eq!(info.total_count, 8); + assert_eq!(info.features.len(), 8); assert!(feature_names.contains(&"metrics-gpu")); assert!(feature_names.contains(&"io-scheduler-debug")); assert!(feature_names.contains(&"manual-test-runners")); assert!(!feature_names.contains(&"metrics")); + assert!(!feature_names.contains(&"direct-io")); } #[test] @@ -942,7 +935,7 @@ mod tests { assert!(output.contains("| metrics-gpu |")); assert!(output.contains("| io-scheduler-debug |")); assert!(output.contains("| manual-test-runners |")); - assert!(output.contains("| direct-io | enabled by default |")); - assert!(output.contains("| full | metrics-gpu + ftps + swift + webdav + direct-io |")); + assert!(output.contains("| full | metrics-gpu + ftps + swift + webdav |")); + assert!(!output.contains("| direct-io |")); } } diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index fe406ab1f..1ec82aef8 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -53,8 +53,7 @@ use rustfs_heal::{ create_ahm_services_cancel_token, heal::storage::ECStoreHealStorage, init_heal_manager, shutdown_ahm_services, }; use rustfs_iam::{init_iam_sys, init_oidc_sys}; -use rustfs_metrics::init_metrics_system; -use rustfs_obs::{init_obs, set_global_guard}; +use rustfs_obs::{init_metrics_runtime, init_obs, set_global_guard}; use rustfs_scanner::init_data_scanner; use rustfs_utils::{ ExternalEnvCompatReport, apply_external_env_compat, get_env_bool_with_aliases, net::parse_and_resolve_address, @@ -563,7 +562,7 @@ async fn run(config: rustfs::config::Config) -> Result<()> { if rustfs_obs::observability_metric_enabled() { // Initialize metrics system - init_metrics_system(ctx.clone()); + init_metrics_runtime(ctx.clone()); // Initialize auto-tuner for performance optimization (optional) rustfs::init::init_auto_tuner(ctx.clone()).await; diff --git a/scripts/check_metrics_migration_refs.sh b/scripts/check_metrics_migration_refs.sh new file mode 100755 index 000000000..e77ceadd5 --- /dev/null +++ b/scripts/check_metrics_migration_refs.sh @@ -0,0 +1,73 @@ +#!/usr/bin/env bash + +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" + +# Allowed references during migration bootstrap (T00 baseline). +# Keep entries minimal and remove them as callsites are migrated. +ALLOWLIST=( + "crates/obs/src/lib.rs" + "crates/obs/src/metrics/mod.rs" + "crates/obs/src/metrics/scheduler.rs" +) + +is_allowed_path() { + local path="$1" + local allow + for allow in "${ALLOWLIST[@]}"; do + if [[ "$path" == "$allow" ]]; then + return 0 + fi + done + + return 1 +} + +MATCHES=() +while IFS= read -r line; do + MATCHES+=("$line") +done < <( + cd "$ROOT_DIR" + rg -n --no-heading \ + -e 'rustfs_metrics::' \ + -e '\binit_metrics_system\b' \ + -e '\binit_metrics_collectors\b' \ + rustfs/src crates \ + --glob '**/*.rs' \ + --glob '!**/tests/**' \ + --glob '!docs/**' || true +) + +while IFS= read -r line; do + MATCHES+=("$line") +done < <( + cd "$ROOT_DIR" + rg -n --no-heading \ + -e 'rustfs-metrics' \ + -e 'crates/metrics' \ + Cargo.toml rustfs crates \ + --glob '**/*.toml' \ + --glob '**/*.rs' \ + --glob '!**/tests/**' \ + --glob '!docs/**' || true +) + +VIOLATIONS=() + +for hit in "${MATCHES[@]}"; do + file="${hit%%:*}" + if is_allowed_path "$file"; then + continue + fi + + VIOLATIONS+=("$hit") +done + +if (( ${#VIOLATIONS[@]} > 0 )); then + echo "Metrics migration reference guard failed: found non-allowlisted references" + printf '%s\n' "${VIOLATIONS[@]}" + exit 1 +fi + +echo "Metrics migration reference guard passed."