From 4d682cc5777810d5d4f987fe5039da549b23bc67 Mon Sep 17 00:00:00 2001 From: Chris Date: Tue, 15 Sep 2026 01:06:29 +0800 Subject: [PATCH] feat(connect): execute performance network jobs (#7877) --- .../v1/fixtures/heartbeat/MANIFEST.sha256 | 2 +- .../heartbeat/producer-capabilities.json | 2 + protocol/agent/v1/fixtures/manifest.sha256 | 2 +- rustfs/src/connect/diagnostics/job.rs | 185 +++++++++++++++++- rustfs/src/connect/diagnostics/mod.rs | 2 + .../src/connect/diagnostics/perf_network.rs | 39 ++++ rustfs/src/connect/diagnostics/profile_cpu.rs | 8 + 7 files changed, 234 insertions(+), 6 deletions(-) diff --git a/protocol/agent/v1/fixtures/heartbeat/MANIFEST.sha256 b/protocol/agent/v1/fixtures/heartbeat/MANIFEST.sha256 index 9de2fce00..f5271d17f 100644 --- a/protocol/agent/v1/fixtures/heartbeat/MANIFEST.sha256 +++ b/protocol/agent/v1/fixtures/heartbeat/MANIFEST.sha256 @@ -1,6 +1,6 @@ 975c1ca53eefeef6766a6fc0b3d3281f7408255342b0686e5e2aee5ad055414c duplicate.json 963529a38a02849c6c2acc6d72668dca9f63218b49c89fae41a451b584850411 overflow.json -d8cf9de12c459aff83736d761ba003ca2544e45477fbbc5d1c6ae43a1e78a482 producer-capabilities.json +6afec3806c98fb55d477455b43935798d8fdb7f9ae39bd9f6637465539e5c02b producer-capabilities.json e3adeee1c8a19aa17e70894896fb79c072e3785bea3611b93c11e79f039ed5af stale.json 22cc7337f545271ebf11ca6b76e7e479d62c51278ac001a8d870b07d0fa9884e unknown.json b06a72ff1d964efe65e996797705ec2ef72574a813aec5d8e9aff26c899f1352 valid.json diff --git a/protocol/agent/v1/fixtures/heartbeat/producer-capabilities.json b/protocol/agent/v1/fixtures/heartbeat/producer-capabilities.json index 6b9c1033c..716d651f2 100644 --- a/protocol/agent/v1/fixtures/heartbeat/producer-capabilities.json +++ b/protocol/agent/v1/fixtures/heartbeat/producer-capabilities.json @@ -11,6 +11,7 @@ "producerCapabilities": [ "performance.client@1", "performance.drive@1", + "performance.network@1", "performance.object@1", "performance.siteReplication@1", "logs.capture@1", @@ -33,6 +34,7 @@ "inventory.environment@1", "performance.client@1", "performance.drive@1", + "performance.network@1", "performance.object@1", "performance.siteReplication@1", "logs.capture@1", diff --git a/protocol/agent/v1/fixtures/manifest.sha256 b/protocol/agent/v1/fixtures/manifest.sha256 index cd0e69cb2..4ffe6e680 100644 --- a/protocol/agent/v1/fixtures/manifest.sha256 +++ b/protocol/agent/v1/fixtures/manifest.sha256 @@ -1,7 +1,7 @@ a499742c06046e9cf2f781fc17d8b4cf95e018ab45fa41ee5944ba15b0e2902c auth 9b54e39584a442270ae486e8378df1d6f168c45d0c59baba9459b46bb212d2df version 159d116e965b51771f1687fc6ec9bdfbf2a4a0e6d6aa26d3f1046b3dac2697dd registration -5905ef6b39f02d5846912878c2762537e54875ef1a8666a2ee2be17b489e9f68 heartbeat +fe1287eac16832bed88320254245ba07a985e6ae7e66e1d1570e324ffa972873 heartbeat cc6c3f3bf5e6a938f2e00b8ce42d8fa7bfb974f4eb2927101d4ede4ebdc10dd5 inventory 0207458ec9b97b368e2f347da112511a3aabd27d752890cacda4fa2408e7f19e object-performance a464d4c695d76dd985c0f2abf68c2135c080221098ad2427022f85243a7bf6ae network-performance diff --git a/rustfs/src/connect/diagnostics/job.rs b/rustfs/src/connect/diagnostics/job.rs index f7ac16ea5..34229ad9f 100644 --- a/rustfs/src/connect/diagnostics/job.rs +++ b/rustfs/src/connect/diagnostics/job.rs @@ -30,15 +30,18 @@ use tokio_util::sync::CancellationToken; use uuid::{Uuid, Variant, Version}; use super::{ - CPU_PROFILE_CAPABILITY, LocalProfileConsent, LocalTopConsent, MAX_TOP_EXPORT_VALIDITY, 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, sign_top_export, + CPU_PROFILE_CAPABILITY, 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_network, runtime_network_peer_aliases, sign_network_export, sign_top_export, }; use crate::connect::DeviceIdentity; const PROTOCOL_VERSION: &str = "v1"; const PROFILE_CPU_JOB_TYPE: &str = "profile.cpu"; +const PERFORMANCE_NETWORK_JOB_TYPE: &str = "performance.network"; const TOP_API_JOB_TYPE: &str = "top.api"; const TOP_LOCKS_JOB_TYPE: &str = "top.locks"; pub const DIAGNOSTIC_JOB_SIGNATURE_DOMAIN: &[u8] = b"rustfs-connect-agent-job-v1\0"; @@ -47,6 +50,8 @@ const MAX_FUTURE_SKEW_SECONDS: i64 = 300; const MAX_OUTPUT_BYTES: u64 = 524_288; const MAX_MEMORY_BYTES: u64 = 64 * 1024 * 1024; const MAX_CPU_MILLIS: u64 = 30_000; +const MAX_NETWORK_CPU_MILLIS: u64 = 5_000; +const MIN_NETWORK_MEMORY_BYTES: u64 = 1_048_576; const MAX_TOP_API_CPU_MILLIS: u64 = 5_000; const MIN_TOP_API_MEMORY_BYTES: u64 = 1_048_576; const MAX_TOP_LOCKS_CPU_MILLIS: u64 = 5_000; @@ -55,6 +60,7 @@ const MIN_TOP_LOCKS_MEMORY_BYTES: u64 = 1_048_576; #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum DiagnosticJobKind { ProfileCpu, + PerformanceNetwork, TopApi, TopLocks, } @@ -84,6 +90,8 @@ pub struct DiagnosticJobParameters { pub consent_expires_at: String, pub duration_millis: u64, pub sample_period_micros: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub traffic_bytes: Option, } #[derive(Clone, Debug, PartialEq, Eq, Deserialize, Serialize)] @@ -389,6 +397,17 @@ impl DiagnosticJobEnvelope { { return Err(DiagnosticJobError::LimitExceeded); } + match (kind, self.parameters.traffic_bytes) { + (DiagnosticJobKind::PerformanceNetwork, Some(1..=MAX_NETWORK_TRAFFIC_BYTES)) => { + if self.limits.max_cpu_millis > MAX_NETWORK_CPU_MILLIS || self.limits.max_memory_bytes < MIN_NETWORK_MEMORY_BYTES + { + return Err(DiagnosticJobError::LimitExceeded); + } + } + (DiagnosticJobKind::PerformanceNetwork, _) => return Err(DiagnosticJobError::LimitExceeded), + (_, None) => {} + (_, Some(_)) => return Err(DiagnosticJobError::Invalid), + } if kind == DiagnosticJobKind::TopLocks && (self.limits.max_cpu_millis > MAX_TOP_LOCKS_CPU_MILLIS || self.limits.max_memory_bytes < MIN_TOP_LOCKS_MEMORY_BYTES) @@ -403,6 +422,9 @@ impl DiagnosticJobEnvelope { (PROFILE_CPU_JOB_TYPE, [capability], PROFILE_SCHEMA_VERSION) if capability == CPU_PROFILE_CAPABILITY => { Ok(DiagnosticJobKind::ProfileCpu) } + (PERFORMANCE_NETWORK_JOB_TYPE, [capability], NETWORK_SCHEMA_VERSION) if capability == NETWORK_CAPABILITY => { + Ok(DiagnosticJobKind::PerformanceNetwork) + } (TOP_API_JOB_TYPE, [capability], version) if capability == TOP_API_CAPABILITY && version == u16::from(TOP_SCHEMA_VERSION) => { @@ -431,11 +453,102 @@ 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::PerformanceNetwork => { + execute_performance_network_job(envelope, nonce, identity, provenance, cancel).await + } DiagnosticJobKind::TopApi => execute_top_api_job(envelope, identity, provenance, cancel).await, DiagnosticJobKind::TopLocks => execute_top_locks_job(envelope, identity, provenance, cancel).await, } } +async fn execute_performance_network_job( + envelope: DiagnosticJobEnvelope, + nonce: [u8; 32], + identity: &DeviceIdentity, + provenance: ProfileProvenance, + cancel: &CancellationToken, +) -> Result { + let Some(peer_aliases) = runtime_network_peer_aliases() else { + return Ok(DiagnosticJobExecution { + job_id: envelope.job_id, + outcome: "FAILED".to_owned(), + reason: "SOURCE_UNAVAILABLE".to_owned(), + artifact_uid: None, + artifact_sha256: None, + artifact_bytes: None, + }); + }; + let traffic_bytes = envelope.parameters.traffic_bytes.ok_or(DiagnosticJobError::LimitExceeded)?; + let peer_count = u64::try_from(peer_aliases.len()).map_err(|_| DiagnosticJobError::LimitExceeded)?; + let traffic_bytes_per_peer = traffic_bytes + .checked_div(peer_count) + .filter(|value| *value > 0) + .ok_or(DiagnosticJobError::LimitExceeded)?; + let expire = parse_time(&envelope.expire_time)?; + let consent_expire = parse_time(&envelope.parameters.consent_expires_at)?; + let request = NetworkPerformanceRequest { + 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, + schema_version: envelope.schema_version, + capability: NETWORK_CAPABILITY.to_owned(), + consent: LocalNetworkConsent { + 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), + peer_aliases, + traffic_bytes_per_peer, + provenance: NetworkProvenance::new( + provenance.source_commit(), + provenance.executable_sha256(), + provenance.rustfs_version(), + provenance.build_features().to_vec(), + ), + }; + let measurement = measure_network(&request, cancel).await.map_err(network_capture_failure)?; + let measured_outcome = measurement.result.outcome(); + let measured_reason = measurement.result.reason_code(); + let outcome = measured_outcome.as_str().to_owned(); + let reason = measured_reason.as_str().to_owned(); + if !matches!(measured_outcome, NetworkOutcome::Succeeded | NetworkOutcome::Partial) { + let (outcome, reason) = match (measured_outcome, measured_reason) { + (NetworkOutcome::Unsupported, NetworkReasonCode::SourceUnavailable) => { + ("FAILED".to_owned(), "SOURCE_UNAVAILABLE".to_owned()) + } + (NetworkOutcome::Unsupported, _) => ("FAILED".to_owned(), "COLLECTION_FAILED".to_owned()), + _ => (outcome, reason), + }; + return Ok(DiagnosticJobExecution { + job_id: envelope.job_id, + outcome, + reason, + artifact_uid: None, + artifact_sha256: None, + artifact_bytes: None, + }); + } + let export = sign_network_export(&request, &measurement, identity, cancel).map_err(network_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, + reason, + 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], @@ -636,6 +749,22 @@ fn top_capture_failure(error: super::TopCaptureError) -> DiagnosticJobError { } } +fn network_capture_failure(error: NetworkPerformanceError) -> DiagnosticJobError { + match error { + NetworkPerformanceError::Cancelled => DiagnosticJobError::Cancelled, + NetworkPerformanceError::LimitExceeded | NetworkPerformanceError::Busy => DiagnosticJobError::LimitExceeded, + _ => DiagnosticJobError::CollectionFailed, + } +} + +fn network_export_failure(error: NetworkPerformanceError) -> DiagnosticJobError { + match error { + NetworkPerformanceError::Cancelled => DiagnosticJobError::Cancelled, + NetworkPerformanceError::LimitExceeded => DiagnosticJobError::LimitExceeded, + _ => DiagnosticJobError::ExportFailed, + } +} + fn top_export_failure(error: super::TopCaptureError) -> DiagnosticJobError { match error { super::TopCaptureError::Cancelled => DiagnosticJobError::Cancelled, @@ -703,6 +832,7 @@ mod tests { consent_expires_at: "2030-01-01T00:01:00Z".to_owned(), duration_millis: 1_000, sample_period_micros: 10_000, + traffic_bytes: None, }, signature: DiagnosticJobSignature { algorithm: "Ed25519".to_owned(), @@ -769,6 +899,53 @@ mod tests { ); } + #[test] + fn accepts_only_a_bounded_network_traffic_budget() { + let mut network = envelope(); + network.job_type = PERFORMANCE_NETWORK_JOB_TYPE.to_owned(); + network.required_capabilities = vec![NETWORK_CAPABILITY.to_owned()]; + network.limits.max_cpu_millis = MAX_NETWORK_CPU_MILLIS; + network.parameters.duration_millis = MAX_NETWORK_CPU_MILLIS; + network.parameters.traffic_bytes = Some(MAX_NETWORK_TRAFFIC_BYTES); + let (network, signer) = signed_envelope(network); + signer + .verify(&network, &target(&network), "2030-01-01T00:00:10Z".parse().expect("time")) + .expect("valid performance.network job"); + + let mut missing = network.clone(); + missing.parameters.traffic_bytes = None; + assert_eq!( + signer.verify(&missing, &target(&missing), "2030-01-01T00:00:10Z".parse().expect("time")), + Err(DiagnosticJobError::LimitExceeded) + ); + + let mut zero = network.clone(); + zero.parameters.traffic_bytes = Some(0); + assert_eq!( + signer.verify(&zero, &target(&zero), "2030-01-01T00:00:10Z".parse().expect("time")), + Err(DiagnosticJobError::LimitExceeded) + ); + + let mut unbounded = network.clone(); + unbounded.parameters.traffic_bytes = Some(MAX_NETWORK_TRAFFIC_BYTES + 1); + assert_eq!( + signer.verify(&unbounded, &target(&unbounded), "2030-01-01T00:00:10Z".parse().expect("time")), + Err(DiagnosticJobError::LimitExceeded) + ); + + let mut profile = signed().0; + profile.parameters.traffic_bytes = Some(1); + assert_eq!( + signer.verify(&profile, &target(&profile), "2030-01-01T00:00:10Z".parse().expect("time")), + Err(DiagnosticJobError::Invalid) + ); + + let unsigned_network = serde_json::to_value(network.unsigned()).expect("network envelope"); + let unsigned_profile = serde_json::to_value(envelope().unsigned()).expect("profile envelope"); + assert_eq!(unsigned_network["parameters"]["trafficBytes"], MAX_NETWORK_TRAFFIC_BYTES); + assert!(unsigned_profile["parameters"].get("trafficBytes").is_none()); + } + #[test] fn accepts_only_the_bounded_top_locks_capability_pair() { let mut top = envelope(); diff --git a/rustfs/src/connect/diagnostics/mod.rs b/rustfs/src/connect/diagnostics/mod.rs index f47deae58..af579fdd6 100644 --- a/rustfs/src/connect/diagnostics/mod.rs +++ b/rustfs/src/connect/diagnostics/mod.rs @@ -42,6 +42,7 @@ mod trace_runtime; pub const CONNECT_DIAGNOSTIC_CAPABILITIES: &[&str] = &[ perf_client::CLIENT_CAPABILITY, perf_drive::DRIVE_CAPABILITY, + perf_network::NETWORK_CAPABILITY, perf_object::OBJECT_CAPABILITY, perf_site_replication::SITE_REPLICATION_CAPABILITY, logs::LOGS_CAPABILITY, @@ -88,6 +89,7 @@ pub use perf_drive::{ DriveTargetReasonCode, DriveTargetResult, DriveTargetUnits, LocalDriveConsent, SavedDriveExport, SignedDriveExport, measure_drive, save_signed_drive_export, sign_drive_export, validate_drive_limits, }; +pub(crate) use perf_network::runtime_network_peer_aliases; pub use perf_network::{ LocalNetworkConsent, MAX_ARCHIVE_BYTES as MAX_NETWORK_ARCHIVE_BYTES, MAX_BANDWIDTH_BYTES_PER_SECOND as MAX_NETWORK_BANDWIDTH_BYTES_PER_SECOND, diff --git a/rustfs/src/connect/diagnostics/perf_network.rs b/rustfs/src/connect/diagnostics/perf_network.rs index 94e2e2f65..04e184379 100644 --- a/rustfs/src/connect/diagnostics/perf_network.rs +++ b/rustfs/src/connect/diagnostics/perf_network.rs @@ -71,6 +71,18 @@ pub enum NetworkOutcome { Cancelled, } +impl NetworkOutcome { + pub const fn as_str(self) -> &'static str { + match self { + Self::Succeeded => "SUCCEEDED", + Self::Partial => "PARTIAL", + Self::Failed => "FAILED", + Self::Unsupported => "UNSUPPORTED", + Self::Cancelled => "CANCELLED", + } + } +} + #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] #[serde(rename_all = "SCREAMING_SNAKE_CASE")] pub enum NetworkReasonCode { @@ -86,6 +98,23 @@ pub enum NetworkReasonCode { CollectionFailed, } +impl NetworkReasonCode { + pub const fn as_str(self) -> &'static str { + match self { + Self::Complete => "COMPLETE", + Self::LimitExceeded => "LIMIT_EXCEEDED", + Self::SourceUnavailable => "SOURCE_UNAVAILABLE", + Self::PermissionDenied => "PERMISSION_DENIED", + Self::UnsupportedTool => "UNSUPPORTED_TOOL", + Self::UnsupportedVersion => "UNSUPPORTED_VERSION", + Self::UnsupportedPlatform => "UNSUPPORTED_PLATFORM", + Self::Cancelled => "CANCELLED", + Self::InvalidInput => "INVALID_INPUT", + Self::CollectionFailed => "COLLECTION_FAILED", + } + } +} + #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] #[serde(rename_all = "SCREAMING_SNAKE_CASE")] pub enum PeerReasonCode { @@ -390,6 +419,16 @@ pub async fn measure_network( measure_network_with_harness(request, &harness, cancel).await } +pub(crate) fn runtime_network_peer_aliases() -> Option> { + let endpoint_pools = crate::runtime_sources::current_endpoints_handle()?; + let aliases = NetworkPeerProbeClient::from_endpoint_pools(&endpoint_pools) + .targets() + .into_iter() + .map(|target| target.alias) + .collect::>(); + (!aliases.is_empty()).then_some(aliases) +} + pub async fn measure_network_with_harness( request: &NetworkPerformanceRequest, harness: &dyn NetworkPeerHarness, diff --git a/rustfs/src/connect/diagnostics/profile_cpu.rs b/rustfs/src/connect/diagnostics/profile_cpu.rs index 6139e92a5..ac16e25bd 100644 --- a/rustfs/src/connect/diagnostics/profile_cpu.rs +++ b/rustfs/src/connect/diagnostics/profile_cpu.rs @@ -174,6 +174,14 @@ impl ProfileProvenance { &self.executable_sha256 } + pub(crate) fn source_commit(&self) -> &str { + &self.source_commit + } + + pub(crate) fn rustfs_version(&self) -> &str { + &self.rustfs_version + } + pub(crate) fn build_features(&self) -> &[String] { &self.build_features }