From 1cd9d1ed5dfd0edd1305801994f88859aa182005 Mon Sep 17 00:00:00 2001 From: Chris Date: Mon, 14 Sep 2026 10:00:48 +0800 Subject: [PATCH] feat(connect): collect native Linux thread states (#7806) Collect bounded aggregate thread states from Linux procfs while preserving explicit unsupported outcomes for unavailable scopes. --- rustfs/src/connect/diagnostics/mod.rs | 5 +- rustfs/src/connect/diagnostics/profile_cpu.rs | 41 +++++ .../connect/diagnostics/profile_threads.rs | 174 ++++++++++++++++-- rustfs/src/connect/mod.rs | 17 +- rustfs/src/startup_entrypoint.rs | 4 +- rustfs/tests/connect_profile_threads.rs | 94 ++++++++-- 6 files changed, 295 insertions(+), 40 deletions(-) diff --git a/rustfs/src/connect/diagnostics/mod.rs b/rustfs/src/connect/diagnostics/mod.rs index 920d530f6..81c7c914c 100644 --- a/rustfs/src/connect/diagnostics/mod.rs +++ b/rustfs/src/connect/diagnostics/mod.rs @@ -95,8 +95,9 @@ pub use perf_site_replication::{ pub use profile_cpu::{ CPU_PROFILE_CAPABILITY, LocalProfileConsent, MAX_PROFILE_DURATION, MEMORY_PROFILE_CAPABILITY, PROFILE_SCHEMA_VERSION, ProfileCaptureRequest, ProfileData, ProfileError, ProfileOutcome, ProfileProvenance, ProfileReasonCode, ProfileResult, - ProfileTool, SavedProfileExport, SignedProfileExport, THREAD_PROFILE_CAPABILITY, ThreadProfileScope, capture_cpu_profile, - encode_signed_profile_export, export_cpu_profile, save_signed_profile_export, + ProfileTool, SavedProfileExport, SignedProfileExport, THREAD_PROFILE_CAPABILITY, ThreadProfileData, ThreadProfileScope, + ThreadState, ThreadStateCount, capture_cpu_profile, encode_signed_profile_export, export_cpu_profile, + save_signed_profile_export, }; pub use profile_memory::export_memory_profile; pub use profile_threads::{capture_thread_profile, export_thread_profile}; diff --git a/rustfs/src/connect/diagnostics/profile_cpu.rs b/rustfs/src/connect/diagnostics/profile_cpu.rs index a94f1a04e..4c76877d9 100644 --- a/rustfs/src/connect/diagnostics/profile_cpu.rs +++ b/rustfs/src/connect/diagnostics/profile_cpu.rs @@ -300,6 +300,44 @@ impl MemoryProfileData { } } +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ThreadProfileData { + scope: ThreadProfileScope, + states: Vec, +} + +impl ThreadProfileData { + pub(super) fn native(states: Vec) -> Self { + Self { + scope: ThreadProfileScope::NativeThreads, + states, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum ThreadState { + Runnable, + Waiting, + Blocked, + Unknown, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ThreadStateCount { + state: ThreadState, + thread_count: u64, +} + +impl ThreadStateCount { + pub(super) const fn new(state: ThreadState, thread_count: u64) -> Self { + Self { state, thread_count } + } +} + #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] #[serde(rename_all = "SCREAMING_SNAKE_CASE")] pub enum ThreadProfileScope { @@ -312,6 +350,7 @@ pub enum ThreadProfileScope { pub enum ProfileData { Cpu(CpuProfileData), Memory(MemoryProfileData), + Threads(ThreadProfileData), } #[derive(Clone, Debug, PartialEq, Eq, Serialize)] @@ -416,6 +455,8 @@ pub enum ProfileError { SourceUnavailable, #[error("profile_counter_reset")] CounterReset, + #[error("profile_collection_failed")] + CollectionFailed, #[error("profile_export_signing_failed")] Signing, #[error("profile_export_exists")] diff --git a/rustfs/src/connect/diagnostics/profile_threads.rs b/rustfs/src/connect/diagnostics/profile_threads.rs index 1bf405755..1cfb5607e 100644 --- a/rustfs/src/connect/diagnostics/profile_threads.rs +++ b/rustfs/src/connect/diagnostics/profile_threads.rs @@ -12,43 +12,189 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Explicit thread/runtime profile capability result. +//! Bounded thread-state profile collection. //! -//! Dial9 currently exposes session and disk-buffer state, not bounded counts -//! for RUNNABLE, WAITING, BLOCKED, and UNKNOWN threads. Native thread state -//! would require a separate reviewed platform adapter. Neither source is -//! relabelled as the contract's thread data. +//! Linux native collection reads only the state byte from this process's +//! `/proc/self/task/*/stat` records. Thread names, identifiers, stacks, paths, +//! addresses, and raw procfs bytes cannot enter the exported result. Tokio +//! runtime state remains explicitly unsupported because Dial9 does not expose +//! the contract's RUNNABLE, WAITING, BLOCKED, and UNKNOWN counts. +#[cfg(target_os = "linux")] +use std::fs::{self, File}; +#[cfg(target_os = "linux")] +use std::io::{ErrorKind, Read as _}; +#[cfg(target_os = "linux")] +use std::time::Instant; use tokio_util::sync::CancellationToken; +#[cfg(target_os = "linux")] +use super::profile_cpu::{CollectorLease, ProfileData, ThreadProfileData, ThreadState, ThreadStateCount}; use super::profile_cpu::{ - ProfileCaptureRequest, ProfileError, ProfileReasonCode, ProfileResult, ProfileTool, ThreadProfileScope, check_cancel, - encode_signed_profile_export, unix_now, + ProfileCaptureRequest, ProfileError, ProfileReasonCode, ProfileResult, ProfileTool, SignedProfileExport, ThreadProfileScope, + check_cancel, encode_signed_profile_export, unix_now, }; use crate::connect::DeviceIdentity; -use super::profile_cpu::SignedProfileExport; +#[cfg(target_os = "linux")] +const PROC_TASK_DIRECTORY: &str = "/proc/self/task"; +#[cfg(target_os = "linux")] +const MAX_NATIVE_THREADS: usize = 4_096; +#[cfg(target_os = "linux")] +const MAX_PROC_STAT_BYTES: u64 = 4_096; pub fn capture_thread_profile( request: &ProfileCaptureRequest, - _scope: ThreadProfileScope, + scope: ThreadProfileScope, cancel: &CancellationToken, ) -> Result { request.validate(ProfileTool::Threads, unix_now()?)?; check_cancel(cancel)?; - Ok(ProfileResult::unsupported( + + if scope == ThreadProfileScope::TokioRuntime { + return Ok(ProfileResult::unsupported( + request, + ProfileTool::Threads, + ProfileReasonCode::UnsupportedTool, + )); + } + + #[cfg(not(target_os = "linux"))] + return Ok(ProfileResult::unsupported( request, ProfileTool::Threads, - ProfileReasonCode::UnsupportedTool, - )) + ProfileReasonCode::UnsupportedPlatform, + )); + + #[cfg(target_os = "linux")] + { + let _lease = CollectorLease::acquire()?; + let started = Instant::now(); + let deadline = started.checked_add(request.duration).ok_or(ProfileError::LimitExceeded)?; + let data = collect_native_thread_states(cancel, deadline)?; + check_cancel(cancel)?; + return Ok(ProfileResult::succeeded( + request, + ProfileTool::Threads, + started.elapsed(), + ProfileData::Threads(data), + )); + } } -pub fn export_thread_profile( +pub async fn export_thread_profile( request: &ProfileCaptureRequest, scope: ThreadProfileScope, key: &DeviceIdentity, cancel: &CancellationToken, ) -> Result { - let result = capture_thread_profile(request, scope, cancel)?; + let owned_request = request.clone(); + let owned_cancel = cancel.clone(); + let result = tokio::task::spawn_blocking(move || capture_thread_profile(&owned_request, scope, &owned_cancel)) + .await + .map_err(|_| ProfileError::CollectionFailed)??; encode_signed_profile_export(request, &result, key, cancel) } + +#[cfg(target_os = "linux")] +fn collect_native_thread_states(cancel: &CancellationToken, deadline: Instant) -> Result { + let entries = fs::read_dir(PROC_TASK_DIRECTORY).map_err(|_| ProfileError::SourceUnavailable)?; + let mut counts = [0_u64; 4]; + let mut visited = 0_usize; + + for entry in entries { + check_cancel(cancel)?; + if Instant::now() >= deadline { + return Err(ProfileError::TimedOut); + } + let entry = entry.map_err(|_| ProfileError::SourceUnavailable)?; + let name = entry.file_name(); + let Some(name) = name.to_str() else { + return Err(ProfileError::SourceUnavailable); + }; + if name.is_empty() || !name.bytes().all(|byte| byte.is_ascii_digit()) { + return Err(ProfileError::SourceUnavailable); + } + visited = visited.checked_add(1).ok_or(ProfileError::LimitExceeded)?; + if visited > MAX_NATIVE_THREADS { + return Err(ProfileError::LimitExceeded); + } + + let state = match read_proc_stat_state(&entry.path().join("stat"))? { + Some(state) => state, + None => continue, + }; + let index = match state { + ThreadState::Runnable => 0, + ThreadState::Waiting => 1, + ThreadState::Blocked => 2, + ThreadState::Unknown => 3, + }; + counts[index] = counts[index].checked_add(1).ok_or(ProfileError::LimitExceeded)?; + } + + if counts.iter().all(|count| *count == 0) { + return Err(ProfileError::SourceUnavailable); + } + + Ok(ThreadProfileData::native(vec![ + ThreadStateCount::new(ThreadState::Runnable, counts[0]), + ThreadStateCount::new(ThreadState::Waiting, counts[1]), + ThreadStateCount::new(ThreadState::Blocked, counts[2]), + ThreadStateCount::new(ThreadState::Unknown, counts[3]), + ])) +} + +#[cfg(target_os = "linux")] +fn read_proc_stat_state(path: &std::path::Path) -> Result, ProfileError> { + let file = match File::open(path) { + Ok(file) => file, + Err(error) if error.kind() == ErrorKind::NotFound => return Ok(None), + Err(_) => return Err(ProfileError::SourceUnavailable), + }; + let mut bytes = Vec::with_capacity(256); + file.take(MAX_PROC_STAT_BYTES + 1) + .read_to_end(&mut bytes) + .map_err(|_| ProfileError::SourceUnavailable)?; + if bytes.is_empty() || bytes.len() as u64 > MAX_PROC_STAT_BYTES { + return Err(ProfileError::SourceUnavailable); + } + parse_proc_stat_state(&bytes).map(Some) +} + +#[cfg(target_os = "linux")] +fn parse_proc_stat_state(stat: &[u8]) -> Result { + let closing = stat + .iter() + .rposition(|byte| *byte == b')') + .ok_or(ProfileError::SourceUnavailable)?; + let suffix = stat.get(closing + 1..).ok_or(ProfileError::SourceUnavailable)?; + let state = match suffix { + [b' ', state, b' ', ..] => *state, + _ => return Err(ProfileError::SourceUnavailable), + }; + Ok(match state { + b'R' => ThreadState::Runnable, + b'D' => ThreadState::Blocked, + b'S' | b'I' | b'T' | b't' | b'W' => ThreadState::Waiting, + _ => ThreadState::Unknown, + }) +} + +#[cfg(all(test, target_os = "linux"))] +mod tests { + use super::*; + + #[test] + fn proc_stat_parser_uses_only_the_kernel_state_byte() { + for (raw, expected) in [ + (b"123 (worker secret path) R 1 2".as_slice(), ThreadState::Runnable), + (b"123 (worker) S 1 2", ThreadState::Waiting), + (b"123 (worker) D 1 2", ThreadState::Blocked), + (b"123 (worker) Z 1 2", ThreadState::Unknown), + ] { + assert_eq!(parse_proc_stat_state(raw).expect("valid proc stat"), expected); + } + assert!(matches!(parse_proc_stat_state(b"123 malformed"), Err(ProfileError::SourceUnavailable))); + } +} diff --git a/rustfs/src/connect/mod.rs b/rustfs/src/connect/mod.rs index d28a1e2f0..b86b311ca 100644 --- a/rustfs/src/connect/mod.rs +++ b/rustfs/src/connect/mod.rs @@ -70,14 +70,15 @@ pub use diagnostics::{ TELEMETRY_OTLP_CAPABILITY, TELEMETRY_RECORD_CAPABILITY, TELEMETRY_REPLAY_CAPABILITY, TELEMETRY_SCHEMA_VERSION, THREAD_PROFILE_CAPABILITY, TelemetryArtifactConsent, TelemetryArtifactError, TelemetryArtifactRequest, TelemetryCoverage, TelemetryDiagnosticResult, TelemetryOperation, TelemetryOutcome, TelemetryProducerError, TelemetryProvenance, - TelemetryReasonCode, TelemetrySpan, TelemetrySpanStatus, TelemetryTool, ThreadProfileScope, TraceAnalysis, - TraceAnalysisError, TraceRecordCapture, TraceRecordCompletion, TraceRecordLimits, TraceReplayError, analyze_trace, - capture_cpu_profile, capture_thread_profile, encode_signed_profile_export, encode_signed_telemetry_export, - export_cpu_profile, export_logs, export_memory_profile, export_thread_profile, export_trace_otlp, export_trace_otlp_result, - measure_client, measure_drive, read_protected_client_credential, record_diagnostic_result, record_trace, record_trace_bus, - replay_trace, replay_trace_result, run_local_environment_once, save_signed_client_export, save_signed_drive_export, - save_signed_log_export, save_signed_profile_export, save_signed_telemetry_export, sign_client_export, sign_drive_export, - spawn_environment_schedule, validate_client_limits, validate_drive_limits, + TelemetryReasonCode, TelemetrySpan, TelemetrySpanStatus, TelemetryTool, ThreadProfileData, ThreadProfileScope, ThreadState, + ThreadStateCount, TraceAnalysis, TraceAnalysisError, TraceRecordCapture, TraceRecordCompletion, TraceRecordLimits, + TraceReplayError, analyze_trace, capture_cpu_profile, capture_thread_profile, encode_signed_profile_export, + encode_signed_telemetry_export, export_cpu_profile, export_logs, export_memory_profile, export_thread_profile, + export_trace_otlp, export_trace_otlp_result, measure_client, measure_drive, read_protected_client_credential, + record_diagnostic_result, record_trace, record_trace_bus, replay_trace, replay_trace_result, run_local_environment_once, + save_signed_client_export, save_signed_drive_export, save_signed_log_export, save_signed_profile_export, + save_signed_telemetry_export, sign_client_export, sign_drive_export, spawn_environment_schedule, validate_client_limits, + validate_drive_limits, }; pub use diagnostics::{ LocalNetworkConsent, MAX_NETWORK_ARCHIVE_BYTES, MAX_NETWORK_BANDWIDTH_BYTES_PER_SECOND, MAX_NETWORK_DECOMPRESSED_BYTES, diff --git a/rustfs/src/startup_entrypoint.rs b/rustfs/src/startup_entrypoint.rs index 54037998b..de281cdc0 100644 --- a/rustfs/src/startup_entrypoint.rs +++ b/rustfs/src/startup_entrypoint.rs @@ -1250,7 +1250,9 @@ async fn execute_connect_profile(options: ConnectProfileOpts) -> Result<()> { Some(ConnectThreadProfileScope::NativeThreads) => ThreadProfileScope::NativeThreads, None => return Err(Error::other("--thread-scope is required for the threads profile")), }; - export_thread_profile(&request, scope, &key, &cancel).map_err(Error::other) + export_thread_profile(&request, scope, &key, &cancel) + .await + .map_err(Error::other) } } }; diff --git a/rustfs/tests/connect_profile_threads.rs b/rustfs/tests/connect_profile_threads.rs index be87cdc61..5e35d6b91 100644 --- a/rustfs/tests/connect_profile_threads.rs +++ b/rustfs/tests/connect_profile_threads.rs @@ -29,7 +29,11 @@ use profile_cpu::{ THREAD_PROFILE_CAPABILITY, ThreadProfileScope, }; use profile_threads::{capture_thread_profile, export_thread_profile}; +#[cfg(target_os = "linux")] +use std::io::{Cursor, Read as _}; use tokio_util::sync::CancellationToken; +#[cfg(target_os = "linux")] +use zip::ZipArchive; fn request() -> ProfileCaptureRequest { let now = SystemTime::now().duration_since(UNIX_EPOCH).expect("current time").as_secs() as i64; @@ -58,24 +62,84 @@ fn request() -> ProfileCaptureRequest { } } -#[test] -fn tokio_and_native_thread_scopes_remain_explicitly_unsupported() { +#[tokio::test] +async fn tokio_thread_scope_remains_explicitly_unsupported() { let key = connect::DeviceIdentity::generate(); - for scope in [ThreadProfileScope::TokioRuntime, ThreadProfileScope::NativeThreads] { - let result = capture_thread_profile(&request(), scope, &CancellationToken::new()).expect("unsupported result"); - assert_eq!(result.outcome(), ProfileOutcome::Unsupported); - assert_eq!(result.reason_code(), ProfileReasonCode::UnsupportedTool); - assert!(result.data().is_none(), "unsupported scope must not publish zero state counts"); - let json = serde_json::to_value(result).expect("result JSON"); - assert_eq!(json["toolId"], "profile.threads"); - assert_eq!(json["capability"], "profile.threads@1"); - assert!(json["data"].is_null()); + let result = capture_thread_profile(&request(), ThreadProfileScope::TokioRuntime, &CancellationToken::new()) + .expect("unsupported result"); + assert_eq!(result.outcome(), ProfileOutcome::Unsupported); + assert_eq!(result.reason_code(), ProfileReasonCode::UnsupportedTool); + assert!(result.data().is_none(), "unsupported scope must not publish zero state counts"); + let json = serde_json::to_value(result).expect("result JSON"); + assert_eq!(json["toolId"], "profile.threads"); + assert_eq!(json["capability"], "profile.threads@1"); + assert!(json["data"].is_null()); - let export = export_thread_profile(&request(), scope, &key, &CancellationToken::new()).expect("unsupported export"); - assert_eq!(export.tool.id(), "profile.threads"); - assert_eq!(export.outcome, ProfileOutcome::Unsupported); - assert_eq!(export.reason_code, ProfileReasonCode::UnsupportedTool); + let export = export_thread_profile(&request(), ThreadProfileScope::TokioRuntime, &key, &CancellationToken::new()) + .await + .expect("unsupported export"); + assert_eq!(export.tool.id(), "profile.threads"); + assert_eq!(export.outcome, ProfileOutcome::Unsupported); + assert_eq!(export.reason_code, ProfileReasonCode::UnsupportedTool); +} + +#[cfg(target_os = "linux")] +#[tokio::test] +async fn native_thread_scope_exports_bounded_redacted_state_counts() { + let key = connect::DeviceIdentity::generate(); + let result = capture_thread_profile(&request(), ThreadProfileScope::NativeThreads, &CancellationToken::new()) + .expect("native thread result"); + assert_eq!(result.outcome(), ProfileOutcome::Succeeded); + assert_eq!(result.reason_code(), ProfileReasonCode::Complete); + let json = serde_json::to_value(result).expect("result JSON"); + assert_eq!(json["data"]["scope"], "NATIVE_THREADS"); + let states = json["data"]["states"].as_array().expect("thread states"); + assert_eq!(states.len(), 4); + assert!(states.iter().all(|state| state["threadCount"].as_u64().is_some())); + assert!(states.iter().map(|state| state["threadCount"].as_u64().unwrap()).sum::() > 0); + let encoded = serde_json::to_string(&json).expect("encoded result"); + for forbidden in ["/proc/", "task/", "worker", "secret", "stack", "address", "threadId"] { + assert!(!encoded.contains(forbidden), "result leaked forbidden material: {forbidden}"); } + + let export = export_thread_profile(&request(), ThreadProfileScope::NativeThreads, &key, &CancellationToken::new()) + .await + .expect("native thread export"); + assert_eq!(export.outcome, ProfileOutcome::Succeeded); + assert!(export.archive_bytes.len() <= profile_cpu::MAX_ARCHIVE_BYTES); + let mut archive = ZipArchive::new(Cursor::new(export.archive_bytes)).expect("profile archive"); + let mut result = String::new(); + archive + .by_name("result.json") + .expect("profile result") + .read_to_string(&mut result) + .expect("read profile result"); + assert!(result.contains("\"scope\":\"NATIVE_THREADS\"")); + for forbidden in ["/proc/", "task/", "worker", "secret", "stack", "address", "threadId"] { + assert!(!result.contains(forbidden), "archive leaked forbidden material: {forbidden}"); + } +} + +#[cfg(target_os = "linux")] +#[test] +fn native_thread_scope_honors_the_monotonic_deadline() { + let mut expired = request(); + expired.duration = Duration::from_nanos(1); + expired.sample_period = Duration::from_nanos(1); + assert!(matches!( + capture_thread_profile(&expired, ThreadProfileScope::NativeThreads, &CancellationToken::new()), + Err(ProfileError::TimedOut) + )); +} + +#[cfg(not(target_os = "linux"))] +#[test] +fn native_thread_scope_is_explicitly_unsupported_off_linux() { + let result = capture_thread_profile(&request(), ThreadProfileScope::NativeThreads, &CancellationToken::new()) + .expect("unsupported result"); + assert_eq!(result.outcome(), ProfileOutcome::Unsupported); + assert_eq!(result.reason_code(), ProfileReasonCode::UnsupportedPlatform); + assert!(result.data().is_none()); } #[test]