From bdd18f19964ed5c1eb12b48542da1606694e8cd6 Mon Sep 17 00:00:00 2001 From: Chris Date: Tue, 15 Sep 2026 01:23:42 +0800 Subject: [PATCH] feat(connect): execute thread profile service jobs (#7880) --- rustfs/src/connect/diagnostics/job.rs | 131 +++++++++++++++++++++++++- 1 file changed, 127 insertions(+), 4 deletions(-) diff --git a/rustfs/src/connect/diagnostics/job.rs b/rustfs/src/connect/diagnostics/job.rs index 8e427d4f2..eb52bdfc2 100644 --- a/rustfs/src/connect/diagnostics/job.rs +++ b/rustfs/src/connect/diagnostics/job.rs @@ -34,15 +34,17 @@ use super::{ DriveProvenance, LocalDriveConsent, LocalNetworkConsent, LocalProfileConsent, LocalTopConsent, MAX_NETWORK_TRAFFIC_BYTES, MAX_TOP_EXPORT_VALIDITY, NETWORK_CAPABILITY, NETWORK_SCHEMA_VERSION, NetworkOutcome, NetworkPerformanceError, NetworkPerformanceRequest, NetworkProvenance, NetworkReasonCode, PROFILE_SCHEMA_VERSION, ProfileCaptureRequest, - ProfileProvenance, TOP_API_CAPABILITY, TOP_CLASSIFICATION, TOP_LOCKS_CAPABILITY, TOP_SCHEMA_VERSION, TopApiOperation, - TopCaptureLimits, TopCaptureRequest, TopCaptureScope, TopOutcome, capture_cpu_profile, capture_top_api, capture_top_locks, - encode_signed_profile_export, measure_drive, measure_network, runtime_network_peer_aliases, sign_drive_export, - sign_network_export, sign_top_export, sign_top_export_with_nonce, + ProfileOutcome, ProfileProvenance, THREAD_PROFILE_CAPABILITY, TOP_API_CAPABILITY, TOP_CLASSIFICATION, TOP_LOCKS_CAPABILITY, + TOP_SCHEMA_VERSION, ThreadProfileScope, TopApiOperation, TopCaptureLimits, TopCaptureRequest, TopCaptureScope, TopOutcome, + capture_cpu_profile, capture_thread_profile, capture_top_api, capture_top_locks, encode_signed_profile_export, measure_drive, + measure_network, runtime_network_peer_aliases, sign_drive_export, sign_network_export, sign_top_export, + sign_top_export_with_nonce, }; use crate::connect::DeviceIdentity; const PROTOCOL_VERSION: &str = "v1"; const PROFILE_CPU_JOB_TYPE: &str = "profile.cpu"; +const PROFILE_THREADS_JOB_TYPE: &str = "profile.threads"; const PERFORMANCE_DRIVE_JOB_TYPE: &str = "performance.drive"; const PERFORMANCE_NETWORK_JOB_TYPE: &str = "performance.network"; const TOP_API_JOB_TYPE: &str = "top.api"; @@ -69,6 +71,7 @@ const MIN_TOP_LOCKS_MEMORY_BYTES: u64 = 1_048_576; #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum DiagnosticJobKind { ProfileCpu, + ProfileThreads, PerformanceDrive, PerformanceNetwork, TopApi, @@ -459,6 +462,9 @@ impl DiagnosticJobEnvelope { (PROFILE_CPU_JOB_TYPE, [capability], PROFILE_SCHEMA_VERSION) if capability == CPU_PROFILE_CAPABILITY => { Ok(DiagnosticJobKind::ProfileCpu) } + (PROFILE_THREADS_JOB_TYPE, [capability], PROFILE_SCHEMA_VERSION) if capability == THREAD_PROFILE_CAPABILITY => { + Ok(DiagnosticJobKind::ProfileThreads) + } (PERFORMANCE_DRIVE_JOB_TYPE, [capability], DRIVE_SCHEMA_VERSION) if capability == DRIVE_CAPABILITY => { Ok(DiagnosticJobKind::PerformanceDrive) } @@ -493,6 +499,7 @@ pub async fn execute_diagnostic_job( let envelope = job.envelope; match envelope.kind()? { DiagnosticJobKind::ProfileCpu => execute_profile_cpu_job(envelope, nonce, identity, provenance, cancel).await, + DiagnosticJobKind::ProfileThreads => execute_profile_threads_job(envelope, nonce, identity, provenance, cancel).await, DiagnosticJobKind::PerformanceDrive => { let Some(scratch_root) = runtime_drive_scratch_root() else { return Ok(failed_drive_execution(&envelope.job_id, "SOURCE_UNAVAILABLE")); @@ -716,6 +723,68 @@ async fn execute_performance_network_job( }) } +async fn execute_profile_threads_job( + envelope: DiagnosticJobEnvelope, + nonce: [u8; 32], + identity: &DeviceIdentity, + provenance: ProfileProvenance, + cancel: &CancellationToken, +) -> Result { + let expire = parse_time(&envelope.expire_time)?; + let consent_expire = parse_time(&envelope.parameters.consent_expires_at)?; + let request = ProfileCaptureRequest { + organization_name: envelope.organization_name, + cluster_name: envelope.cluster_name, + device_name: envelope.device_name, + run_uid: envelope.job_id.clone(), + artifact_uid: envelope.parameters.artifact_uid.clone(), + schema_version: envelope.schema_version, + capability: THREAD_PROFILE_CAPABILITY.to_owned(), + consent: LocalProfileConsent { + consent_uid: envelope.parameters.consent_uid, + policy_revision: envelope.parameters.consent_policy_revision, + expires_at_unix: consent_expire.timestamp(), + confirmed: true, + }, + produced_at_unix: Utc::now().timestamp(), + expires_at_unix: expire.timestamp(), + nonce, + duration: Duration::from_millis(envelope.parameters.duration_millis), + sample_period: Duration::from_micros(envelope.parameters.sample_period_micros), + provenance, + }; + let owned_request = request.clone(); + let owned_cancel = cancel.clone(); + let result = tokio::task::spawn_blocking(move || { + capture_thread_profile(&owned_request, ThreadProfileScope::NativeThreads, &owned_cancel) + }) + .await + .map_err(|_| DiagnosticJobError::CollectionFailed)? + .map_err(capture_failure)?; + if result.outcome() == ProfileOutcome::Unsupported { + return Ok(DiagnosticJobExecution { + job_id: envelope.job_id, + outcome: result.outcome().as_str().to_owned(), + reason: result.reason_code().as_str().to_owned(), + artifact_uid: None, + artifact_sha256: None, + artifact_bytes: None, + }); + } + let export = encode_signed_profile_export(&request, &result, identity, cancel).map_err(export_failure)?; + if export.archive_bytes.len() > usize::try_from(envelope.limits.max_output_bytes).unwrap_or(usize::MAX) { + return Err(DiagnosticJobError::LimitExceeded); + } + Ok(DiagnosticJobExecution { + job_id: envelope.job_id, + outcome: result.outcome().as_str().to_owned(), + reason: result.reason_code().as_str().to_owned(), + artifact_uid: Some(export.artifact_uid), + artifact_sha256: Some(export.archive_sha256), + artifact_bytes: Some(export.archive_bytes), + }) +} + async fn execute_profile_cpu_job( envelope: DiagnosticJobEnvelope, nonce: [u8; 32], @@ -1061,6 +1130,60 @@ mod tests { .expect("valid job"); } + #[test] + fn accepts_only_the_thread_profile_capability_pair() { + let mut threads = envelope(); + threads.job_type = PROFILE_THREADS_JOB_TYPE.to_owned(); + threads.required_capabilities = vec![THREAD_PROFILE_CAPABILITY.to_owned()]; + let (threads, signer) = signed_envelope(threads); + signer + .verify(&threads, &target(&threads), "2030-01-01T00:00:10Z".parse().expect("time")) + .expect("valid profile.threads job"); + + let mut mismatched = threads; + mismatched.required_capabilities = vec![CPU_PROFILE_CAPABILITY.to_owned()]; + assert_eq!( + signer.verify(&mismatched, &target(&mismatched), "2030-01-01T00:00:10Z".parse().expect("time")), + Err(DiagnosticJobError::Unsupported) + ); + } + + #[tokio::test] + async fn executes_thread_profile_jobs_against_the_service_process() { + let now = Utc::now(); + let mut envelope = envelope(); + envelope.job_type = PROFILE_THREADS_JOB_TYPE.to_owned(); + envelope.required_capabilities = vec![THREAD_PROFILE_CAPABILITY.to_owned()]; + envelope.create_time = now.to_rfc3339_opts(chrono::SecondsFormat::Secs, true); + envelope.expire_time = (now + chrono::Duration::seconds(30)).to_rfc3339_opts(chrono::SecondsFormat::Secs, true); + envelope.parameters.consent_expires_at = + (now + chrono::Duration::seconds(60)).to_rfc3339_opts(chrono::SecondsFormat::Secs, true); + let execution = execute_diagnostic_job( + VerifiedDiagnosticJob { + envelope, + nonce: [7_u8; 32], + }, + &DeviceIdentity::generate(), + ProfileProvenance::new("a".repeat(40), "b".repeat(64), "1.0.0", vec![]), + &CancellationToken::new(), + ) + .await + .expect("native thread profile job should execute"); + + #[cfg(target_os = "linux")] + { + assert_eq!(execution.outcome, "SUCCEEDED"); + assert_eq!(execution.reason, "COMPLETE"); + assert!(execution.artifact_bytes.is_some_and(|bytes| !bytes.is_empty())); + } + #[cfg(not(target_os = "linux"))] + { + assert_eq!(execution.outcome, "UNSUPPORTED"); + assert_eq!(execution.reason, "UNSUPPORTED_PLATFORM"); + assert!(execution.artifact_bytes.is_none()); + } + } + #[test] fn accepts_only_the_bounded_top_api_capability_pair() { let mut top = envelope();