From 5851d9eb54dd920d93e70d41383bd7a377e17ba1 Mon Sep 17 00:00:00 2001 From: Chris Date: Wed, 30 Sep 2026 01:06:24 +0800 Subject: [PATCH] fix: filter CPU profile samples to reviewed executable symbols (#8248) --- Cargo.lock | 2 + rustfs/Cargo.toml | 6 +- rustfs/src/connect/diagnostics/profile_cpu.rs | 122 ++++++++++++++++-- 3 files changed, 121 insertions(+), 9 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index cb80e1077..519567e1d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9553,6 +9553,7 @@ dependencies = [ "metrics-util", "mime_guess", "moka", + "object 0.40.0", "opentelemetry 0.33.0", "opentelemetry-proto", "opentelemetry_sdk 0.33.0", @@ -9632,6 +9633,7 @@ dependencies = [ "snap", "socket2", "subtle", + "symbolic-demangle", "sysinfo", "temp-env", "tempfile", diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 11bd1c8e0..06028243e 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -74,7 +74,7 @@ connect-e2e-short-credentials = [] # Builds the dedicated rustfs-cli-e2e target with a build-time public enrollment root. offline-enrollment-e2e-root = [] rio-v2 = ["rustfs-ecstore/rio-v2"] -pyroscope = ["rustfs-obs/pyroscope", "dep:pyroscope"] +pyroscope = ["rustfs-obs/pyroscope", "dep:pyroscope", "dep:object", "dep:symbolic-demangle"] # Tokio runtime telemetry. Requires `--cfg tokio_unstable`; use `make build-profiling`. dial9 = ["rustfs-obs/dial9"] hotpath = [ @@ -379,6 +379,10 @@ libsystemd.workspace = true [target.'cfg(any(all(target_os = "macos", any(target_arch = "x86_64", target_arch = "aarch64")), all(target_os = "linux", target_env = "gnu", any(target_arch = "x86_64", target_arch = "aarch64"))))'.dependencies] pyroscope = { workspace = true, optional = true, features = ["backend-pprof-rs"] } +[target.'cfg(all(target_os = "linux", target_env = "gnu", any(target_arch = "x86_64", target_arch = "aarch64")))'.dependencies] +object = { workspace = true, optional = true, features = ["read", "std", "elf"] } +symbolic-demangle = { version = "=13.9.0", default-features = false, features = ["rust"], optional = true } + [dev-dependencies] uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] } serial_test = { workspace = true } diff --git a/rustfs/src/connect/diagnostics/profile_cpu.rs b/rustfs/src/connect/diagnostics/profile_cpu.rs index 9b75c3357..420678ead 100644 --- a/rustfs/src/connect/diagnostics/profile_cpu.rs +++ b/rustfs/src/connect/diagnostics/profile_cpu.rs @@ -649,9 +649,12 @@ pub async fn export_cpu_profile( ) ))] mod local_cpu { - use std::collections::HashMap; + use std::collections::{HashMap, HashSet}; use std::time::{Duration, Instant}; + #[cfg(all(target_os = "linux", target_env = "gnu"))] + use object::{Object as _, ObjectSymbol as _}; + use pyroscope::backend::{BackendConfig, PprofConfig, ReportData, pprof_backend}; use sha2::{Digest as _, Sha256}; use tokio_util::sync::CancellationToken; @@ -665,6 +668,10 @@ mod local_cpu { const MAX_UNIQUE_SYMBOLS: usize = 4_096; const MAX_OUTPUT_SYMBOLS: usize = 256; const MAX_SYMBOL_BYTES: usize = 4_096; + #[cfg(all(target_os = "linux", target_env = "gnu"))] + const MAX_CATALOG_BYTES: usize = 32 * 1024 * 1024; + #[cfg(all(target_os = "linux", target_env = "gnu"))] + const MAX_CATALOG_SYMBOLS: usize = 200_000; pub(super) fn collect( request: &ProfileCaptureRequest, @@ -679,6 +686,8 @@ mod local_cpu { return Err(ProfileError::LimitExceeded); } + let reviewed_symbols = reviewed_executable_symbols(cancel)?; + let started = Instant::now(); let deadline = started.checked_add(request.duration).ok_or(ProfileError::LimitExceeded)?; let mut backend = pprof_backend(PprofConfig { sample_rate }, BackendConfig::default()) @@ -694,7 +703,7 @@ mod local_cpu { return Err(ProfileError::SourceUnavailable); }; - let mut accumulator = Accumulator::new(request.nonce); + let mut accumulator = Accumulator::new(request.nonce, reviewed_symbols.as_ref()); for report in reports { for (stack, count) in report.data { accumulator.record_stack(stack.frames.iter().filter_map(|frame| frame.name.as_deref()), count)?; @@ -718,18 +727,89 @@ mod local_cpu { } } - struct Accumulator { + // The packaged Connect catalog is derived from this executable's ELF text symbols. + // Match that exact symbol set before signing; shared-library and runtime frames + // cannot be reviewed by a catalog bound to the RustFS executable alone. + fn reviewed_executable_symbols(cancel: &CancellationToken) -> Result>, ProfileError> { + #[cfg(all(target_os = "linux", target_env = "gnu"))] + { + return load_executable_symbols(cancel).map(Some); + } + #[cfg(not(all(target_os = "linux", target_env = "gnu")))] + { + check_cancel(cancel)?; + Ok(None) + } + } + + #[cfg(all(target_os = "linux", target_env = "gnu"))] + fn load_executable_symbols(cancel: &CancellationToken) -> Result, ProfileError> { + check_cancel(cancel)?; + let path = std::env::current_exe().map_err(|_| ProfileError::SourceUnavailable)?; + let executable = std::fs::File::open(path).map_err(|_| ProfileError::SourceUnavailable)?; + let size = executable.metadata().map_err(|_| ProfileError::SourceUnavailable)?.len(); + if size == 0 || size > 1_073_741_824 { + return Err(ProfileError::SourceUnavailable); + } + let cache = object::read::ReadCache::new(executable); + let file = object::File::parse(&cache).map_err(|_| ProfileError::SourceUnavailable)?; + if file.format() != object::BinaryFormat::Elf || file.symbol_table().is_none() { + return Err(ProfileError::SourceUnavailable); + } + let mut symbols = HashSet::new(); + let mut symbol_bytes = 0_usize; + for (index, symbol) in file.symbols().chain(file.dynamic_symbols()).enumerate() { + if index % 1_024 == 0 { + check_cancel(cancel)?; + } + if symbol.kind() != object::SymbolKind::Text || symbol.is_undefined() { + continue; + } + if let Some(name) = symbol.name_bytes().ok().and_then(catalog_symbol_name) + && !symbols.contains(&name) + { + symbol_bytes = symbol_bytes + .checked_add(name.len() + 3) + .ok_or(ProfileError::SourceUnavailable)?; + if symbol_bytes > MAX_CATALOG_BYTES || symbols.len() >= MAX_CATALOG_SYMBOLS { + return Err(ProfileError::SourceUnavailable); + } + symbols.insert(name); + } + } + if symbols.is_empty() { + return Err(ProfileError::SourceUnavailable); + } + Ok(symbols) + } + + #[cfg(all(target_os = "linux", target_env = "gnu"))] + fn catalog_symbol_name(raw: &[u8]) -> Option { + let name = symbolic_demangle::demangle(&String::from_utf8_lossy(raw)).into_owned(); + if name.is_empty() + || name.len() > MAX_SYMBOL_BYTES + || matches!(name.as_str(), "Unknown" | "") + || name.bytes().any(|byte| byte < 32 || byte == 127) + { + return None; + } + Some(name) + } + + struct Accumulator<'a> { nonce: [u8; 32], + reviewed_symbols: Option<&'a HashSet>, samples: HashMap, stack_records: usize, accepted_sample_count: u64, dropped_sample_count: u64, } - impl Accumulator { - fn new(nonce: [u8; 32]) -> Self { + impl<'a> Accumulator<'a> { + fn new(nonce: [u8; 32], reviewed_symbols: Option<&'a HashSet>) -> Self { Self { nonce, + reviewed_symbols, samples: HashMap::new(), stack_records: 0, accepted_sample_count: 0, @@ -756,6 +836,9 @@ mod local_cpu { .into_iter() .find(|symbol| !symbol.is_empty() && symbol.len() <= MAX_SYMBOL_BYTES) .unwrap_or(""); + if self.reviewed_symbols.is_some_and(|reviewed| !reviewed.contains(symbol)) { + return self.drop_samples(accepted_count); + } let symbol_id = symbol_id(&self.nonce, symbol); if !self.samples.contains_key(&symbol_id) && self.samples.len() >= MAX_UNIQUE_SYMBOLS { return self.drop_samples(accepted_count); @@ -820,12 +903,12 @@ mod local_cpu { #[test] fn summary_uses_nonce_bound_ids_and_excludes_raw_symbols() { let raw_symbol = "rustfs::storage::disk::read_object"; - let mut first = Accumulator::new([7; 32]); + let mut first = Accumulator::new([7; 32], None); first .record_stack([raw_symbol].into_iter(), 9) .expect("first stack should be recorded"); let first = first.finish(10_000).expect("first summary should be produced"); - let mut second = Accumulator::new([8; 32]); + let mut second = Accumulator::new([8; 32], None); second .record_stack([raw_symbol].into_iter(), 9) .expect("second stack should be recorded"); @@ -841,7 +924,7 @@ mod local_cpu { #[test] fn summary_bounds_samples_and_output_symbols() { - let mut accumulator = Accumulator::new([3; 32]); + let mut accumulator = Accumulator::new([3; 32], None); for index in 0..=MAX_OUTPUT_SYMBOLS { let symbol = format!("rustfs::bounded::{index}"); accumulator @@ -867,6 +950,29 @@ mod local_cpu { Err(ProfileError::Cancelled) )); } + + #[test] + fn unreviewed_symbol_counts_are_dropped_without_exporting_their_ids() { + let reviewed = HashSet::from(["rustfs::known".to_owned()]); + let mut accumulator = Accumulator::new([4; 32], Some(&reviewed)); + accumulator + .record_stack(["rustfs::known"].into_iter(), 3) + .expect("reviewed stack"); + accumulator + .record_stack(["libc::unreviewed"].into_iter(), 2) + .expect("unreviewed stack"); + let summary = accumulator.finish(10_000).expect("summary"); + assert_eq!(summary.samples.len(), 1); + assert_eq!(summary.samples[0].sample_count, 3); + assert_eq!(summary.dropped_sample_count, 2); + assert_ne!(summary.samples[0].symbol_id, symbol_id(&[4; 32], "libc::unreviewed")); + + let mut only_unreviewed = Accumulator::new([4; 32], Some(&reviewed)); + only_unreviewed + .record_stack(["libc::unreviewed"].into_iter(), 2) + .expect("unreviewed stack"); + assert!(matches!(only_unreviewed.finish(10_000), Err(ProfileError::SourceUnavailable))); + } } }