gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-18 08:35:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(connect): execute performance network jobs (#7877)

Browse Source
This commit is contained in:
Chris
2026-09-15 01:06:29 +08:00
committed by GitHub
parent c525904731
commit 4d682cc577
7 changed files with 234 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
protocol/agent/v1/fixtures/heartbeat/MANIFEST.sha256
+1 -1
View File
@@ -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
protocol/agent/v1/fixtures/heartbeat/producer-capabilities.json
+2
View File
@@ -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",
protocol/agent/v1/fixtures/manifest.sha256
+1 -1
View File
@@ -1,7 +1,7 @@
a499742c06046e9cf2f781fc17d8b4cf95e018ab45fa41ee5944ba15b0e2902c auth
9b54e39584a442270ae486e8378df1d6f168c45d0c59baba9459b46bb212d2df version
159d116e965b51771f1687fc6ec9bdfbf2a4a0e6d6aa26d3f1046b3dac2697dd registration
5905ef6b39f02d5846912878c2762537e54875ef1a8666a2ee2be17b489e9f68 heartbeat
fe1287eac16832bed88320254245ba07a985e6ae7e66e1d1570e324ffa972873 heartbeat
cc6c3f3bf5e6a938f2e00b8ce42d8fa7bfb974f4eb2927101d4ede4ebdc10dd5 inventory
0207458ec9b97b368e2f347da112511a3aabd27d752890cacda4fa2408e7f19e object-performance
a464d4c695d76dd985c0f2abf68c2135c080221098ad2427022f85243a7bf6ae network-performance
rustfs/src/connect/diagnostics/job.rs
+181 -4
View File
@@ -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<u64>,
}
#[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<DiagnosticJobExecution, DiagnosticJobError> {
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();
rustfs/src/connect/diagnostics/mod.rs
+2
View File
@@ -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,
rustfs/src/connect/diagnostics/perf_network.rs
+39
View File
@@ -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<Vec<String>> {
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::<Vec<_>>();
(!aliases.is_empty()).then_some(aliases)
}
pub async fn measure_network_with_harness(
request: &NetworkPerformanceRequest,
harness: &dyn NetworkPeerHarness,
rustfs/src/connect/diagnostics/profile_cpu.rs
+8
View File
@@ -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
}