feat(connect): add site replication performance producer (#7769)

feat(connect): add site replication performance core
2026-09-19 10:03:13 +08:00 · 2026-09-14 01:42:53 +08:00
parent 1066a08aca
commit c1190c24e6
11 changed files with 1925 additions and 6 deletions
@@ -5,7 +5,7 @@ f707b09b257c7d652eff2a3f680ecb0c8c45ebb4c2defa8dc07f9dd7ee46ca04  heartbeat
 cc6c3f3bf5e6a938f2e00b8ce42d8fa7bfb974f4eb2927101d4ede4ebdc10dd5  inventory
 0207458ec9b97b368e2f347da112511a3aabd27d752890cacda4fa2408e7f19e  object-performance
 a464d4c695d76dd985c0f2abf68c2135c080221098ad2427022f85243a7bf6ae  network-performance
-0dc06ec2caa8764a51d44e4176959462aa5d0ed9fe5a38453a681ed9d83b8ad6  site-replication-performance
+813cbbf209e8fc9ddaafaeb4eb4745024b1c0edcb26634073a85023844141477  site-replication-performance
 fad21eec58d4d4d547893af9050cdc5eaf9d4f84b15fb423214b16be50e7ff7a  telemetry
 484da2e7599148d36e1ac19d0ae5d305c4a96da7a00495a8ea202d3bae240a88  diagnostic-scheduler
 a2557ab1f7f70fb86a8affc8451596c272478fd2d5500c367b8f9e8385d72578  offline-enrollment
@@ -1,2 +1,2 @@
-e77c27f98e932881c316f08e3bea1b6b20d90e460bb1bbd159bfade1cd4e1377  accept-vectors.json
-e06cdc22f043131452bff32fb5b96cf57183c0b14d72c4710e82971364d596d7  reject-vectors.json
+392877f31c52cb762963e9fae32300eb84e44ed695b8915f66241fc319304bcc  accept-vectors.json
+bf58aa86a7a075935c491cb88fc92e4f9611020a1a9ae27af87e3c55d82df86c  reject-vectors.json
@@ -5,7 +5,7 @@
  "schemaVersion": 1,
  "capability": "performance.siteReplication@1",
  "classification": "L2",
-  "description": "Contract-only performance.siteReplication@1 result vectors. They do not enable or advertise a producer. Results contain only aggregate destination-confirmed measurements; endpoint URLs, credentials, CA material or paths, scratch bucket names, object/version identifiers, and cleanup controls are forbidden.",
+  "description": "Performance.siteReplication@1 result vectors for the bounded native producer. Results contain only aggregate destination-confirmed measurements; endpoint URLs, credentials, CA material or paths, scratch bucket names, object/version identifiers, and cleanup controls are forbidden.",
  "fixture": "accept-vectors",
  "vectors": [
    {
@@ -5,7 +5,7 @@
  "schemaVersion": 1,
  "capability": "performance.siteReplication@1",
  "classification": "L2",
-  "description": "Contract-only performance.siteReplication@1 result vectors. They do not enable or advertise a producer. Results contain only aggregate destination-confirmed measurements; endpoint URLs, credentials, CA material or paths, scratch bucket names, object/version identifiers, and cleanup controls are forbidden.",
+  "description": "Performance.siteReplication@1 result vectors for the bounded native producer. Results contain only aggregate destination-confirmed measurements; endpoint URLs, credentials, CA material or paths, scratch bucket names, object/version identifiers, and cleanup controls are forbidden.",
  "fixture": "reject-vectors",
  "vectors": [
    {
@@ -250,10 +250,112 @@ pub enum ConnectPerformanceCommands {
    Client(Box<ConnectClientPerformanceOpts>),
    /// Measure bounded S3 object throughput in a dedicated temporary namespace
    Object(Box<ConnectObjectPerformanceOpts>),
+    /// Measure bounded destination-confirmed site-replication performance
+    SiteReplication(Box<ConnectSiteReplicationPerformanceOpts>),
    /// Measure generated-file write and warm page-cache read performance
    Drive(Box<ConnectDrivePerformanceOpts>),
 }

+#[derive(Args, Clone)]
+pub struct ConnectSiteReplicationPerformanceOpts {
+    /// Directory containing an enrolled Connect device identity
+    #[arg(long = "state-dir")]
+    pub state_dir: PathBuf,
+    /// Source RustFS deployment endpoint
+    #[arg(long = "source-endpoint", value_parser = NonEmptyStringValueParser::new())]
+    pub source_endpoint: String,
+    /// Destination RustFS deployment endpoint
+    #[arg(long = "destination-endpoint", value_parser = NonEmptyStringValueParser::new())]
+    pub destination_endpoint: String,
+    /// Optional PEM root certificate for the source endpoint
+    #[arg(long = "source-ca-file")]
+    pub source_ca_file: Option<PathBuf>,
+    /// Optional PEM root certificate for the destination endpoint
+    #[arg(long = "destination-ca-file")]
+    pub destination_ca_file: Option<PathBuf>,
+    /// Owner-readable file containing the source S3 access key
+    #[arg(long = "source-access-key-file")]
+    pub source_access_key_file: PathBuf,
+    /// Owner-readable file containing the source S3 secret key
+    #[arg(long = "source-secret-key-file")]
+    pub source_secret_key_file: PathBuf,
+    /// Optional owner-readable file containing a source S3 session token
+    #[arg(long = "source-session-token-file")]
+    pub source_session_token_file: Option<PathBuf>,
+    /// Owner-readable file containing the destination S3 access key
+    #[arg(long = "destination-access-key-file")]
+    pub destination_access_key_file: PathBuf,
+    /// Owner-readable file containing the destination S3 secret key
+    #[arg(long = "destination-secret-key-file")]
+    pub destination_secret_key_file: PathBuf,
+    /// Optional owner-readable file containing a destination S3 session token
+    #[arg(long = "destination-session-token-file")]
+    pub destination_session_token_file: Option<PathBuf>,
+    /// New local archive path; an existing file is never replaced
+    #[arg(long)]
+    pub output: PathBuf,
+    /// Negotiated producer schema version
+    #[arg(long = "schema-version", default_value_t = 1)]
+    pub schema_version: u16,
+    /// Negotiated producer capability
+    #[arg(long, default_value = "performance.siteReplication@1", value_parser = NonEmptyStringValueParser::new())]
+    pub capability: String,
+    /// Organization resource name bound to the export
+    #[arg(long, value_parser = NonEmptyStringValueParser::new())]
+    pub organization: String,
+    /// Cluster resource name bound to the export
+    #[arg(long, value_parser = NonEmptyStringValueParser::new())]
+    pub cluster: String,
+    /// Cluster-device resource name bound to the export
+    #[arg(long, value_parser = NonEmptyStringValueParser::new())]
+    pub device: String,
+    /// UUIDv7 diagnostic run identifier issued by Connect
+    #[arg(long = "run-uid", value_parser = NonEmptyStringValueParser::new())]
+    pub run_uid: String,
+    /// UUIDv7 artifact identifier issued by Connect
+    #[arg(long = "artifact-uid", value_parser = NonEmptyStringValueParser::new())]
+    pub artifact_uid: String,
+    /// UUIDv7 consent identifier issued by Connect
+    #[arg(long = "consent-uid", value_parser = NonEmptyStringValueParser::new())]
+    pub consent_uid: String,
+    /// Consent policy revision bound to this measurement
+    #[arg(long = "policy-revision")]
+    pub policy_revision: u64,
+    /// Consent expiry as UTC Unix seconds
+    #[arg(long = "consent-expires-at")]
+    pub consent_expires_at_unix: i64,
+    /// Artifact expiry as UTC Unix seconds
+    #[arg(long = "expires-at")]
+    pub expires_at_unix: i64,
+    /// Generated transfer size in bytes
+    #[arg(long = "traffic-bytes", default_value_t = 65_536)]
+    pub traffic_bytes: u64,
+    /// Maximum wall-clock duration in milliseconds
+    #[arg(long = "duration-millis", default_value_t = 5_000)]
+    pub duration_millis: u64,
+    /// Stable opaque alias for the source deployment
+    #[arg(long = "source-alias", value_parser = NonEmptyStringValueParser::new())]
+    pub source_alias: String,
+    /// Source deployment identifier from site-replication configuration
+    #[arg(long = "source-deployment-id", value_parser = NonEmptyStringValueParser::new())]
+    pub source_deployment_id: String,
+    /// Stable opaque alias for the destination deployment
+    #[arg(long = "destination-alias", value_parser = NonEmptyStringValueParser::new())]
+    pub destination_alias: String,
+    /// Destination deployment identifier from site-replication configuration
+    #[arg(long = "destination-deployment-id", value_parser = NonEmptyStringValueParser::new())]
+    pub destination_deployment_id: String,
+    /// Existing versioned bucket dedicated to disposable performance objects
+    #[arg(long = "scratch-bucket", value_parser = NonEmptyStringValueParser::new())]
+    pub scratch_bucket: String,
+    /// Bounded cleanup window for versions that arrive after cancellation
+    #[arg(long = "late-arrival-cleanup-millis", default_value_t = 1_000)]
+    pub late_arrival_cleanup_millis: u64,
+    /// Confirm this explicit local L2 diagnostic operation
+    #[arg(long = "acknowledge-l2", required = true, action = clap::ArgAction::SetTrue)]
+    pub acknowledge_l2: bool,
+}
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)]
 pub enum ConnectClientPerformanceOperation {
    Get,
@@ -1205,6 +1307,8 @@ pub enum CommandResult {
    ConnectClientPerformance(ConnectClientPerformanceOpts),
    /// Consent-bound S3 object performance export
    ConnectObjectPerformance(ConnectObjectPerformanceOpts),
+    /// Consent-bound site-replication performance export
+    ConnectSiteReplicationPerformance(ConnectSiteReplicationPerformanceOpts),
    /// Consent-bound local Connect profile export
    ConnectProfile(ConnectProfileOpts),
    /// Consent-bound local Connect log export
@@ -50,6 +50,7 @@ mod snapshot;
 mod config_test;

 // Re-export public types
+pub use cli::ConnectSiteReplicationPerformanceOpts;
 pub use cli::{CommandResult, InfoOpts, InfoType};
 pub use cli::{
    ConnectClientPerformanceOperation, ConnectClientPerformanceOpts, ConnectDrivePerformanceOpts, ConnectPerformanceCommands,
@@ -150,6 +150,9 @@ impl Opt {
                    ConnectPerformanceCommands::Client(opts) => Ok(CommandResult::ConnectClientPerformance(*opts)),
                    ConnectPerformanceCommands::Drive(opts) => Ok(CommandResult::ConnectDrivePerformance(*opts)),
                    ConnectPerformanceCommands::Object(opts) => Ok(CommandResult::ConnectObjectPerformance(*opts)),
+                    ConnectPerformanceCommands::SiteReplication(opts) => {
+                        Ok(CommandResult::ConnectSiteReplicationPerformance(*opts))
+                    }
                },
                ConnectCommands::Profile(opts) => Ok(CommandResult::ConnectProfile(opts)),
                ConnectCommands::Logs(opts) => Ok(CommandResult::ConnectLogs(opts)),
@@ -18,6 +18,7 @@ mod perf_client;
 mod perf_drive;
 mod perf_network;
 mod perf_object;
+mod perf_site_replication;
 mod profile_cpu;
 mod profile_memory;
 mod profile_threads;
@@ -76,6 +77,16 @@ pub use perf_object::{
    SignedObjectExport, measure_object, read_protected_object_credential, save_signed_object_export, sign_object_export,
    validate_object_limits,
 };
+pub use perf_site_replication::{
+    LocalSiteReplicationConsent, MAX_SITE_REPLICATION_DURATION, MAX_SITE_REPLICATION_TRAFFIC_BYTES, S3SiteReplicationProbe,
+    SITE_REPLICATION_CAPABILITY, SITE_REPLICATION_SCHEMA_VERSION, SITE_REPLICATION_TOOL_ID, SavedSiteReplicationExport,
+    SignedSiteReplicationExport, SiteReplicationDiagnosticResult, SiteReplicationEndpoint, SiteReplicationMeasurement,
+    SiteReplicationOutcome, SiteReplicationPerformanceData, SiteReplicationPerformanceError, SiteReplicationPerformanceRequest,
+    SiteReplicationProbe, SiteReplicationProbeError, SiteReplicationProbeFuture, SiteReplicationProbeMeasurement,
+    SiteReplicationProvenance, SiteReplicationReasonCode, SiteReplicationTargetReasonCode, SiteReplicationTargetResult,
+    measure_site_replication, read_protected_site_replication_credential, save_signed_site_replication_export,
+    sign_site_replication_export, validate_site_replication_limits,
+};
 pub use profile_cpu::{
    CPU_PROFILE_CAPABILITY, LocalProfileConsent, MAX_PROFILE_DURATION, MEMORY_PROFILE_CAPABILITY, PROFILE_SCHEMA_VERSION,
    ProfileCaptureRequest, ProfileData, ProfileError, ProfileOutcome, ProfileProvenance, ProfileReasonCode, ProfileResult,
@@ -93,6 +93,16 @@ pub use diagnostics::{
    SignedObjectExport, measure_object, read_protected_object_credential, save_signed_object_export, sign_object_export,
    validate_object_limits,
 };
+pub use diagnostics::{
+    LocalSiteReplicationConsent, MAX_SITE_REPLICATION_DURATION, MAX_SITE_REPLICATION_TRAFFIC_BYTES, S3SiteReplicationProbe,
+    SITE_REPLICATION_CAPABILITY, SITE_REPLICATION_SCHEMA_VERSION, SITE_REPLICATION_TOOL_ID, SavedSiteReplicationExport,
+    SignedSiteReplicationExport, SiteReplicationDiagnosticResult, SiteReplicationEndpoint, SiteReplicationMeasurement,
+    SiteReplicationOutcome, SiteReplicationPerformanceData, SiteReplicationPerformanceError, SiteReplicationPerformanceRequest,
+    SiteReplicationProbe, SiteReplicationProbeError, SiteReplicationProbeFuture, SiteReplicationProbeMeasurement,
+    SiteReplicationProvenance, SiteReplicationReasonCode, SiteReplicationTargetReasonCode, SiteReplicationTargetResult,
+    measure_site_replication, read_protected_site_replication_credential, save_signed_site_replication_export,
+    sign_site_replication_export, validate_site_replication_limits,
+};
 pub use diagnostics::{
    LocalTopConsent, MAX_TOP_DURATION, MAX_TOP_EXPORT_VALIDITY, NetworkCounterSnapshot, SavedTopExport, SignedTopExport,
    TOP_CLASSIFICATION, TOP_SCHEMA_VERSION, TopApiData, TopApiOperation, TopCaptureError, TopCaptureLimits, TopCaptureRequest,
@@ -17,7 +17,8 @@ use crate::{
        CommandResult, Config, ConnectClientPerformanceOperation, ConnectClientPerformanceOpts, ConnectDrivePerformanceOpts,
        ConnectEnvironmentInventoryOpts, ConnectLicenseCommands, ConnectLicenseScopeOpts, ConnectLogsMode, ConnectLogsOpts,
        ConnectObjectPerformanceOperation, ConnectObjectPerformanceOpts, ConnectProfileOpts, ConnectProfileTool,
-        ConnectTelemetryArtifactOpts, ConnectTelemetryCommands, ConnectThreadProfileScope, ConnectTopCommands, Opt,
+        ConnectSiteReplicationPerformanceOpts, ConnectTelemetryArtifactOpts, ConnectTelemetryCommands, ConnectThreadProfileScope,
+        ConnectTopCommands, Opt,
    },
    startup_lifecycle::{StartupRuntimeLifecycle, run_startup_runtime_lifecycle},
    startup_preflight::{StartupServerPreflightError, bootstrap_external_prefix_compat, init_startup_server_preflight},
@@ -141,6 +142,9 @@ async fn async_main() -> Result<()> {
        CommandResult::ConnectClientPerformance(options) => return execute_connect_client_performance(options).await,
        CommandResult::ConnectDrivePerformance(options) => return execute_connect_drive_performance(options).await,
        CommandResult::ConnectObjectPerformance(options) => return execute_connect_object_performance(options).await,
+        CommandResult::ConnectSiteReplicationPerformance(options) => {
+            return execute_connect_site_replication_performance(options).await;
+        }
        CommandResult::ConnectProfile(options) => return execute_connect_profile(options).await,
        CommandResult::ConnectLogs(options) => return execute_connect_logs(options).await,
        CommandResult::ConnectTelemetry(command) => return execute_connect_telemetry(command).await,
@@ -909,6 +913,167 @@ async fn execute_connect_object_performance(options: ConnectObjectPerformanceOpt
    Ok(())
 }

+async fn execute_connect_site_replication_performance(options: ConnectSiteReplicationPerformanceOpts) -> Result<()> {
+    use crate::connect::{
+        IdentityStore, LocalSiteReplicationConsent, S3SiteReplicationProbe, SiteReplicationEndpoint, SiteReplicationOutcome,
+        SiteReplicationPerformanceRequest, SiteReplicationProvenance, measure_site_replication,
+        read_protected_site_replication_credential, save_signed_site_replication_export, sign_site_replication_export,
+        validate_site_replication_limits,
+    };
+    use rand::{TryRng as _, rngs::SysRng};
+    use zeroize::Zeroizing;
+
+    let duration = Duration::from_millis(options.duration_millis);
+    let late_arrival_cleanup = Duration::from_millis(options.late_arrival_cleanup_millis);
+    validate_site_replication_limits(duration, options.traffic_bytes, late_arrival_cleanup).map_err(Error::other)?;
+    let key = IdentityStore::new(options.state_dir.join("identity"))
+        .load()
+        .map_err(Error::other)?
+        .ok_or_else(|| Error::other("connect site-replication performance requires an enrolled device identity"))?;
+    let source_access_key = read_protected_site_replication_credential(&options.source_access_key_file).map_err(Error::other)?;
+    let source_secret_key = read_protected_site_replication_credential(&options.source_secret_key_file).map_err(Error::other)?;
+    let source_session_token = options
+        .source_session_token_file
+        .as_deref()
+        .map(read_protected_site_replication_credential)
+        .transpose()
+        .map_err(Error::other)?
+        .unwrap_or_else(|| Zeroizing::new(String::new()));
+    let destination_access_key =
+        read_protected_site_replication_credential(&options.destination_access_key_file).map_err(Error::other)?;
+    let destination_secret_key =
+        read_protected_site_replication_credential(&options.destination_secret_key_file).map_err(Error::other)?;
+    let destination_session_token = options
+        .destination_session_token_file
+        .as_deref()
+        .map(read_protected_site_replication_credential)
+        .transpose()
+        .map_err(Error::other)?
+        .unwrap_or_else(|| Zeroizing::new(String::new()));
+    let source_ca = read_optional_root_ca(options.source_ca_file.as_deref(), "source")?;
+    let destination_ca = read_optional_root_ca(options.destination_ca_file.as_deref(), "destination")?;
+    let source = SiteReplicationEndpoint::new(
+        options.source_alias.clone(),
+        options.source_deployment_id.clone(),
+        &options.source_endpoint,
+        source_ca.as_deref(),
+        source_access_key,
+        source_secret_key,
+        source_session_token,
+        duration,
+    )
+    .map_err(Error::other)?;
+    let destination = SiteReplicationEndpoint::new(
+        options.destination_alias.clone(),
+        options.destination_deployment_id.clone(),
+        &options.destination_endpoint,
+        destination_ca.as_deref(),
+        destination_access_key,
+        destination_secret_key,
+        destination_session_token,
+        duration,
+    )
+    .map_err(Error::other)?;
+    let probe = S3SiteReplicationProbe::new(source, destination);
+    let executable_sha256 = hash_current_executable()?;
+    let produced_at_unix = unix_now()?;
+    let mut nonce = [0_u8; 32];
+    SysRng.try_fill_bytes(&mut nonce).map_err(Error::other)?;
+    let request = SiteReplicationPerformanceRequest {
+        organization_name: options.organization,
+        cluster_name: options.cluster,
+        device_name: options.device,
+        run_uid: options.run_uid,
+        artifact_uid: options.artifact_uid,
+        schema_version: options.schema_version,
+        capability: options.capability,
+        consent: LocalSiteReplicationConsent {
+            consent_uid: options.consent_uid,
+            policy_revision: options.policy_revision,
+            expires_at_unix: options.consent_expires_at_unix,
+            nonce,
+            confirmed: options.acknowledge_l2,
+        },
+        produced_at_unix,
+        expires_at_unix: options.expires_at_unix,
+        duration,
+        traffic_bytes: options.traffic_bytes,
+        source_alias: options.source_alias,
+        source_deployment_id: options.source_deployment_id,
+        destination_alias: options.destination_alias,
+        destination_deployment_id: options.destination_deployment_id,
+        scratch_bucket: options.scratch_bucket,
+        late_arrival_cleanup,
+        provenance: SiteReplicationProvenance::new(
+            crate::version::build::COMMIT_HASH,
+            executable_sha256,
+            env!("CARGO_PKG_VERSION"),
+            enabled_build_features(),
+        ),
+    };
+    let cancel = CancellationToken::new();
+    let measurement = measure_site_replication(&request, &probe, &cancel);
+    tokio::pin!(measurement);
+    let measurement = tokio::select! {
+        biased;
+        signal = tokio::signal::ctrl_c() => {
+            signal.map_err(Error::other)?;
+            cancel.cancel();
+            measurement.await.map_err(Error::other)?
+        }
+        result = measurement.as_mut() => result.map_err(Error::other)?,
+    };
+    let target_json = serde_json::to_string(&measurement.target).map_err(Error::other)?;
+    println!(
+        "tool=performance.siteReplication outcome={} reason={}",
+        measurement.result.outcome().as_str(),
+        measurement.result.reason_code().as_str()
+    );
+    println!("target={target_json}");
+    std::io::stdout().flush()?;
+
+    // Measurement cancellation has already completed bounded late-arrival
+    // cleanup. Preserve that terminal result in the signed offline artifact.
+    let writer_cancel = CancellationToken::new();
+    let export = sign_site_replication_export(&request, &measurement, &key, &writer_cancel).map_err(Error::other)?;
+    let output = options.output;
+    let save_cancel = writer_cancel.clone();
+    let receipt = tokio::task::spawn_blocking(move || save_signed_site_replication_export(&output, &export, &save_cancel))
+        .await
+        .map_err(Error::other)?
+        .map_err(Error::other)?;
+    println!(
+        "artifact={} bytes={} sha256={}",
+        receipt.artifact_uid, receipt.archive_size_bytes, receipt.archive_sha256
+    );
+    println!("upload=not-performed");
+    if measurement.result.outcome() != SiteReplicationOutcome::Succeeded {
+        return Err(Error::other(format!(
+            "site-replication performance collection ended with {}",
+            measurement.result.outcome().as_str()
+        )));
+    }
+    Ok(())
+}
+
+fn read_optional_root_ca(path: Option<&std::path::Path>, label: &str) -> Result<Option<Vec<u8>>> {
+    let Some(path) = path else {
+        return Ok(None);
+    };
+    const MAX_ROOT_CA_BYTES: u64 = 1_048_576;
+    let mut bytes = Vec::with_capacity(16 * 1024);
+    std::fs::File::open(path)?
+        .take(MAX_ROOT_CA_BYTES + 1)
+        .read_to_end(&mut bytes)?;
+    let max_bytes = usize::try_from(MAX_ROOT_CA_BYTES).map_err(Error::other)?;
+    if bytes.len() > max_bytes {
+        return Err(Error::other(format!(
+            "connect site-replication {label} root CA exceeds the 1048576-byte limit"
+        )));
+    }
+    Ok(Some(bytes))
+}
+
 async fn execute_connect_drive_performance(options: ConnectDrivePerformanceOpts) -> Result<()> {
    use crate::connect::{
        DriveOutcome, DrivePerformanceRequest, DriveProvenance, IdentityStore, LocalDriveConsent, measure_drive,