mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-19 17:15:12 +00:00
Add client-to-deployment performance diagnostics (#7726)
feat: add client performance diagnostics
This commit is contained in:
Chris
@@ -32,4 +32,5 @@ pub use request_signature_v4::sign_v4;
|
||||
pub use request_signature_v4::sign_v4_trailer;
|
||||
pub use request_signature_v4::try_pre_sign_v4;
|
||||
pub use request_signature_v4::try_sign_v4;
|
||||
pub use request_signature_v4::try_sign_v4_headers;
|
||||
pub use request_signature_v4::try_sign_v4_trailer;
|
||||
|
||||
@@ -679,6 +679,19 @@ pub fn try_sign_v4(
|
||||
.map_err(|failure| failure.error)
|
||||
}
|
||||
|
||||
pub fn try_sign_v4_headers(
|
||||
parts: request::Parts,
|
||||
content_len: i64,
|
||||
access_key_id: &str,
|
||||
secret_access_key: &str,
|
||||
session_token: &str,
|
||||
location: &str,
|
||||
) -> SignResult<HeaderMap> {
|
||||
let request = request::Request::from_parts(parts, Body::empty());
|
||||
try_sign_v4(request, content_len, access_key_id, secret_access_key, session_token, location)
|
||||
.map(|request| request.into_parts().0.headers)
|
||||
}
|
||||
|
||||
pub fn sign_v4_trailer(
|
||||
req: request::Request<Body>,
|
||||
access_key_id: &str,
|
||||
|
||||
@@ -53,6 +53,7 @@ const CONTENT_TYPE_NDJSON: &str = "application/x-ndjson";
|
||||
pub(crate) const CLIENT_DEVNULL_MAX_BYTES: u64 = 1024 * 1024 * 1024;
|
||||
pub(crate) const CLIENT_DEVNULL_MAX_DURATION: Duration = Duration::from_secs(30);
|
||||
pub(crate) const CLIENT_DEVNULL_MAX_CONCURRENCY: usize = 4;
|
||||
pub(crate) const CLIENT_DEVNULL_SOURCE_MAX_BYTES: u64 = 1024 * 1024;
|
||||
static CLIENT_DEVNULL_ADMISSION: Semaphore = Semaphore::const_new(CLIENT_DEVNULL_MAX_CONCURRENCY);
|
||||
|
||||
/// Cap on how many locks a single `top/locks` response enumerates, matching the
|
||||
@@ -124,6 +125,11 @@ pub fn register_diagnostics_route(r: &mut S3Router<AdminOperation>) -> std::io::
|
||||
format!("{ADMIN_PREFIX}/v3/speedtest/client/devnull").as_str(),
|
||||
AdminOperation(&SpeedtestClientDevnullHandler {}),
|
||||
)?;
|
||||
r.insert(
|
||||
Method::GET,
|
||||
format!("{ADMIN_PREFIX}/v3/speedtest/client/devnull").as_str(),
|
||||
AdminOperation(&SpeedtestClientSourceHandler {}),
|
||||
)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -940,6 +946,23 @@ impl Operation for SpeedtestHandler {
|
||||
/// number (mirrors MinIO's `ClientDevNull`).
|
||||
pub struct SpeedtestClientDevnullHandler {}
|
||||
|
||||
/// `GET /v3/speedtest/client/devnull?bytes=N` — bounded generated download.
|
||||
pub struct SpeedtestClientSourceHandler {}
|
||||
|
||||
fn client_source_bytes(uri: &Uri) -> S3Result<usize> {
|
||||
let bytes = query_value(uri, "bytes")
|
||||
.and_then(|value| value.parse::<u64>().ok())
|
||||
.ok_or_else(|| s3_error!(InvalidRequest, "client speedtest requires a positive bytes parameter"))?;
|
||||
if bytes == 0 || bytes > CLIENT_DEVNULL_SOURCE_MAX_BYTES {
|
||||
return Err(s3_error!(
|
||||
EntityTooLarge,
|
||||
"client speedtest download exceeds the {}-byte limit",
|
||||
CLIENT_DEVNULL_SOURCE_MAX_BYTES
|
||||
));
|
||||
}
|
||||
usize::try_from(bytes).map_err(|_| s3_error!(EntityTooLarge, "client speedtest download exceeds platform limits"))
|
||||
}
|
||||
|
||||
fn validate_client_devnull_content_length(headers: &HeaderMap) -> S3Result<()> {
|
||||
let Some(content_length) = headers.get(CONTENT_LENGTH) else {
|
||||
return Ok(());
|
||||
@@ -1029,6 +1052,17 @@ impl Operation for SpeedtestClientDevnullHandler {
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl Operation for SpeedtestClientSourceHandler {
|
||||
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
authorize(&req, AdminAction::HealthInfoAdminAction).await?;
|
||||
let bytes = client_source_bytes(&req.uri)?;
|
||||
let _permit = acquire_client_devnull_permit()?;
|
||||
|
||||
Ok(S3Response::new((StatusCode::OK, Body::from(vec![0_u8; bytes]))))
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Query helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -1179,6 +1213,32 @@ mod tests {
|
||||
assert_eq!(total, 4);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn client_source_requires_a_bounded_positive_size() {
|
||||
let valid = format!("/rustfs/admin/v3/speedtest/client/devnull?bytes={CLIENT_DEVNULL_SOURCE_MAX_BYTES}")
|
||||
.parse::<Uri>()
|
||||
.expect("valid URI");
|
||||
assert_eq!(
|
||||
client_source_bytes(&valid).expect("size at the limit should succeed"),
|
||||
usize::try_from(CLIENT_DEVNULL_SOURCE_MAX_BYTES).expect("source limit fits usize")
|
||||
);
|
||||
|
||||
for invalid in [
|
||||
"/rustfs/admin/v3/speedtest/client/devnull",
|
||||
"/rustfs/admin/v3/speedtest/client/devnull?bytes=0",
|
||||
"/rustfs/admin/v3/speedtest/client/devnull?bytes=invalid",
|
||||
] {
|
||||
let uri = invalid.parse::<Uri>().expect("valid URI");
|
||||
assert!(client_source_bytes(&uri).is_err(), "{invalid} must fail");
|
||||
}
|
||||
|
||||
let oversized = format!("/rustfs/admin/v3/speedtest/client/devnull?bytes={}", CLIENT_DEVNULL_SOURCE_MAX_BYTES + 1)
|
||||
.parse::<Uri>()
|
||||
.expect("valid URI");
|
||||
let err = client_source_bytes(&oversized).expect_err("oversized source request must fail");
|
||||
assert_eq!(err.code(), &S3ErrorCode::EntityTooLarge);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn client_devnull_rejects_invalid_content_length() {
|
||||
let mut headers = HeaderMap::new();
|
||||
|
||||
@@ -812,6 +812,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
|
||||
HEALTH_INFO,
|
||||
RouteRiskLevel::High,
|
||||
),
|
||||
admin(
|
||||
HttpMethod::Get,
|
||||
"/rustfs/admin/v3/speedtest/client/devnull",
|
||||
HEALTH_INFO,
|
||||
RouteRiskLevel::High,
|
||||
),
|
||||
admin(HttpMethod::Post, "/rustfs/admin/v4/inspect/archive", INSPECT_DATA, RouteRiskLevel::High),
|
||||
// MinIO-compatible profiling / trace endpoints.
|
||||
admin(HttpMethod::Post, "/rustfs/admin/v3/profiling/start", PROFILING, RouteRiskLevel::High),
|
||||
|
||||
@@ -375,6 +375,7 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
|
||||
admin_route(Method::POST, "/v3/speedtest/net"),
|
||||
admin_route(Method::POST, "/v3/speedtest/site"),
|
||||
admin_route(Method::POST, "/v3/speedtest/client/devnull"),
|
||||
admin_route(Method::GET, "/v3/speedtest/client/devnull"),
|
||||
admin_route(Method::GET, "/debug/tls/status"),
|
||||
admin_route(Method::POST, "/v3/kms/create-key"),
|
||||
admin_route(Method::POST, "/v3/kms/key/create"),
|
||||
|
||||
+110
-1
@@ -213,8 +213,115 @@ pub struct ConnectPerformanceOpts {
|
||||
|
||||
#[derive(Subcommand, Clone)]
|
||||
pub enum ConnectPerformanceCommands {
|
||||
/// Measure bounded client-to-deployment transfer performance
|
||||
Client(Box<ConnectClientPerformanceOpts>),
|
||||
/// Measure generated-file write and warm page-cache read performance
|
||||
Drive(ConnectDrivePerformanceOpts),
|
||||
Drive(Box<ConnectDrivePerformanceOpts>),
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)]
|
||||
pub enum ConnectClientPerformanceOperation {
|
||||
Get,
|
||||
Put,
|
||||
}
|
||||
|
||||
#[derive(Args, Clone)]
|
||||
pub struct ConnectClientPerformanceOpts {
|
||||
/// Directory containing an enrolled Connect device identity
|
||||
#[arg(long = "state-dir")]
|
||||
pub state_dir: PathBuf,
|
||||
|
||||
/// RustFS deployment endpoint
|
||||
#[arg(long, value_parser = NonEmptyStringValueParser::new())]
|
||||
pub endpoint: String,
|
||||
|
||||
/// Optional PEM root certificate for the deployment endpoint
|
||||
#[arg(long = "ca-file")]
|
||||
pub ca_file: Option<PathBuf>,
|
||||
|
||||
/// Optional explicit HTTP(S) proxy without embedded credentials
|
||||
#[arg(long, value_parser = NonEmptyStringValueParser::new())]
|
||||
pub proxy: Option<String>,
|
||||
|
||||
/// Owner-readable file containing the S3 access key
|
||||
#[arg(long = "access-key-file")]
|
||||
pub access_key_file: PathBuf,
|
||||
|
||||
/// Owner-readable file containing the S3 secret key
|
||||
#[arg(long = "secret-key-file")]
|
||||
pub secret_key_file: PathBuf,
|
||||
|
||||
/// Optional owner-readable file containing an S3 session token
|
||||
#[arg(long = "session-token-file")]
|
||||
pub 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.client@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,
|
||||
|
||||
/// Client transfer operation
|
||||
#[arg(long, value_enum)]
|
||||
pub operation: ConnectClientPerformanceOperation,
|
||||
|
||||
/// 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 = 1_000)]
|
||||
pub duration_millis: u64,
|
||||
|
||||
/// Stable opaque alias for the deployment target
|
||||
#[arg(long = "target-alias", default_value = "deployment-1", value_parser = NonEmptyStringValueParser::new())]
|
||||
pub target_alias: String,
|
||||
|
||||
/// Confirm this explicit local L1 diagnostic operation
|
||||
#[arg(long = "acknowledge-l1", required = true, action = clap::ArgAction::SetTrue)]
|
||||
pub acknowledge_l1: bool,
|
||||
}
|
||||
|
||||
#[derive(Args, Clone)]
|
||||
@@ -952,6 +1059,8 @@ pub enum CommandResult {
|
||||
ConnectLicense(ConnectLicenseCommands),
|
||||
/// Consent-bound local Connect drive performance export
|
||||
ConnectDrivePerformance(ConnectDrivePerformanceOpts),
|
||||
/// Consent-bound client-to-deployment performance export
|
||||
ConnectClientPerformance(ConnectClientPerformanceOpts),
|
||||
/// Consent-bound local Connect profile export
|
||||
ConnectProfile(ConnectProfileOpts),
|
||||
/// Consent-bound local Connect log export
|
||||
|
||||
@@ -51,7 +51,9 @@ mod config_test;
|
||||
|
||||
// Re-export public types
|
||||
pub use cli::{CommandResult, InfoOpts, InfoType};
|
||||
pub use cli::{ConnectDrivePerformanceOpts, ConnectPerformanceCommands};
|
||||
pub use cli::{
|
||||
ConnectClientPerformanceOperation, ConnectClientPerformanceOpts, ConnectDrivePerformanceOpts, ConnectPerformanceCommands,
|
||||
};
|
||||
pub use cli::{ConnectLicenseArtifactOpts, ConnectLicenseCommands, ConnectLicenseScopeOpts};
|
||||
pub use cli::{ConnectLogsMode, ConnectLogsOpts};
|
||||
pub use cli::{ConnectProfileOpts, ConnectProfileTool, ConnectThreadProfileScope};
|
||||
|
||||
@@ -144,7 +144,8 @@ impl Opt {
|
||||
ConnectCommands::Register(opts) => Ok(CommandResult::ConnectRegister(opts)),
|
||||
ConnectCommands::License(opts) => Ok(CommandResult::ConnectLicense(opts.command)),
|
||||
ConnectCommands::Performance(opts) => match opts.command {
|
||||
ConnectPerformanceCommands::Drive(opts) => Ok(CommandResult::ConnectDrivePerformance(opts)),
|
||||
ConnectPerformanceCommands::Client(opts) => Ok(CommandResult::ConnectClientPerformance(*opts)),
|
||||
ConnectPerformanceCommands::Drive(opts) => Ok(CommandResult::ConnectDrivePerformance(*opts)),
|
||||
},
|
||||
ConnectCommands::Profile(opts) => Ok(CommandResult::ConnectProfile(opts)),
|
||||
ConnectCommands::Logs(opts) => Ok(CommandResult::ConnectLogs(opts)),
|
||||
|
||||
@@ -13,6 +13,7 @@
|
||||
// limitations under the License.
|
||||
|
||||
mod logs;
|
||||
mod perf_client;
|
||||
mod perf_drive;
|
||||
mod profile_cpu;
|
||||
mod profile_memory;
|
||||
@@ -32,6 +33,13 @@ pub use logs::{
|
||||
CaptureMode, LOGS_CAPABILITY, LOGS_SCHEMA_VERSION, LocalLogConsent, LogCaptureError, LogCaptureRequest, LogProvenance,
|
||||
SavedLogExport, SignedLogExport, export_logs, save_signed_log_export,
|
||||
};
|
||||
pub use perf_client::{
|
||||
CLIENT_CAPABILITY, CLIENT_SCHEMA_VERSION, ClientDiagnosticResult, ClientMeasurement, ClientOperation, ClientOutcome,
|
||||
ClientPerformanceData, ClientPerformanceError, ClientPerformanceRequest, ClientProbe, ClientProbeError, ClientProbeFuture,
|
||||
ClientProbeMeasurement, ClientProvenance, ClientReasonCode, ClientTargetParameters, ClientTargetReasonCode,
|
||||
ClientTargetResult, ClientTargetUnits, HttpClientProbe, LocalClientConsent, SavedClientExport, SignedClientExport,
|
||||
measure_client, read_protected_client_credential, save_signed_client_export, sign_client_export, validate_client_limits,
|
||||
};
|
||||
pub use perf_drive::{
|
||||
DRIVE_CAPABILITY, DRIVE_SCHEMA_VERSION, DriveDiagnosticResult, DriveMeasurement, DriveOutcome, DrivePerformanceData,
|
||||
DrivePerformanceError, DrivePerformanceRequest, DriveProvenance, DriveReadMode, DriveReasonCode, DriveTargetParameters,
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
+19
-14
@@ -45,17 +45,21 @@ pub use client::{ClientError, ConnectClient, ConnectConfig};
|
||||
pub use config::{HeartbeatConfig, HeartbeatConfigError, HeartbeatSchedule};
|
||||
pub use credential_store::{CredentialStore, DeviceCredential};
|
||||
pub use diagnostics::{
|
||||
CPU_PROFILE_CAPABILITY, CaptureMode, DRIVE_CAPABILITY, DRIVE_SCHEMA_VERSION, DiagnosticCollectionPolicy, DiagnosticReceipt,
|
||||
DiagnosticScheduleError, DiagnosticScheduleRuntime, DiagnosticScheduleStatus, DriveDiagnosticResult, DriveMeasurement,
|
||||
DriveOutcome, DrivePerformanceData, DrivePerformanceError, DrivePerformanceRequest, DriveProvenance, DriveReadMode,
|
||||
DriveReasonCode, DriveTargetParameters, DriveTargetReasonCode, DriveTargetResult, DriveTargetUnits, LOGS_CAPABILITY,
|
||||
LOGS_SCHEMA_VERSION, LocalDriveConsent, LocalLogConsent, LocalOtlpHeaders, LocalProfileConsent, LocalTelemetryConsent,
|
||||
LocallyReviewedTraceArtifact, LogCaptureError, LogCaptureRequest, LogProvenance, MAX_OTLP_BODY_BYTES, MAX_PROFILE_DURATION,
|
||||
MAX_SAFE_INTEGER, MAX_TELEMETRY_DURATION, MAX_TELEMETRY_RESULT_BYTES, MAX_TELEMETRY_SPANS, MEMORY_PROFILE_CAPABILITY,
|
||||
ObservedTelemetrySpan, OperationSummary, OtlpBatch, OtlpForwardError, OtlpReceipt, PROFILE_SCHEMA_VERSION,
|
||||
ProfileCaptureRequest, ProfileData, ProfileError, ProfileOutcome, ProfileProvenance, ProfileReasonCode, ProfileResult,
|
||||
ProfileTool, ReceiptOutcome, RecordedTrace, ReplayedTrace, SavedDriveExport, SavedLogExport, SavedProfileExport,
|
||||
SavedTelemetryExport, SignedDriveExport, SignedLogExport, SignedProfileExport, SignedTelemetryExport,
|
||||
CLIENT_CAPABILITY, CLIENT_SCHEMA_VERSION, CPU_PROFILE_CAPABILITY, CaptureMode, ClientDiagnosticResult, ClientMeasurement,
|
||||
ClientOperation, ClientOutcome, ClientPerformanceData, ClientPerformanceError, ClientPerformanceRequest, ClientProbe,
|
||||
ClientProbeError, ClientProbeFuture, ClientProbeMeasurement, ClientProvenance, ClientReasonCode, ClientTargetParameters,
|
||||
ClientTargetReasonCode, ClientTargetResult, ClientTargetUnits, DRIVE_CAPABILITY, DRIVE_SCHEMA_VERSION,
|
||||
DiagnosticCollectionPolicy, DiagnosticReceipt, DiagnosticScheduleError, DiagnosticScheduleRuntime, DiagnosticScheduleStatus,
|
||||
DriveDiagnosticResult, DriveMeasurement, DriveOutcome, DrivePerformanceData, DrivePerformanceError, DrivePerformanceRequest,
|
||||
DriveProvenance, DriveReadMode, DriveReasonCode, DriveTargetParameters, DriveTargetReasonCode, DriveTargetResult,
|
||||
DriveTargetUnits, HttpClientProbe, LOGS_CAPABILITY, LOGS_SCHEMA_VERSION, LocalClientConsent, LocalDriveConsent,
|
||||
LocalLogConsent, LocalOtlpHeaders, LocalProfileConsent, LocalTelemetryConsent, LocallyReviewedTraceArtifact, LogCaptureError,
|
||||
LogCaptureRequest, LogProvenance, MAX_OTLP_BODY_BYTES, MAX_PROFILE_DURATION, MAX_SAFE_INTEGER, MAX_TELEMETRY_DURATION,
|
||||
MAX_TELEMETRY_RESULT_BYTES, MAX_TELEMETRY_SPANS, MEMORY_PROFILE_CAPABILITY, ObservedTelemetrySpan, OperationSummary,
|
||||
OtlpBatch, OtlpForwardError, OtlpReceipt, PROFILE_SCHEMA_VERSION, ProfileCaptureRequest, ProfileData, ProfileError,
|
||||
ProfileOutcome, ProfileProvenance, ProfileReasonCode, ProfileResult, ProfileTool, ReceiptOutcome, RecordedTrace,
|
||||
ReplayedTrace, SavedClientExport, SavedDriveExport, SavedLogExport, SavedProfileExport, SavedTelemetryExport,
|
||||
SignedClientExport, SignedDriveExport, SignedLogExport, SignedProfileExport, SignedTelemetryExport,
|
||||
TELEMETRY_OTLP_CAPABILITY, TELEMETRY_RECORD_CAPABILITY, TELEMETRY_REPLAY_CAPABILITY, TELEMETRY_SCHEMA_VERSION,
|
||||
THREAD_PROFILE_CAPABILITY, TelemetryArtifactConsent, TelemetryArtifactError, TelemetryArtifactRequest, TelemetryCoverage,
|
||||
TelemetryDiagnosticResult, TelemetryOperation, TelemetryOutcome, TelemetryProducerError, TelemetryProvenance,
|
||||
@@ -63,9 +67,10 @@ pub use diagnostics::{
|
||||
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_drive, record_diagnostic_result, record_trace, record_trace_bus, replay_trace, replay_trace_result,
|
||||
run_local_environment_once, save_signed_drive_export, save_signed_log_export, save_signed_profile_export,
|
||||
save_signed_telemetry_export, sign_drive_export, spawn_environment_schedule, validate_drive_limits,
|
||||
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::{
|
||||
LocalTopConsent, MAX_TOP_DURATION, MAX_TOP_EXPORT_VALIDITY, NetworkCounterSnapshot, SavedTopExport, SignedTopExport,
|
||||
|
||||
@@ -14,9 +14,10 @@
|
||||
|
||||
use crate::{
|
||||
config::{
|
||||
CommandResult, Config, ConnectDrivePerformanceOpts, ConnectLicenseCommands, ConnectLicenseScopeOpts, ConnectLogsMode,
|
||||
ConnectLogsOpts, ConnectProfileOpts, ConnectProfileTool, ConnectTelemetryArtifactOpts, ConnectTelemetryCommands,
|
||||
ConnectThreadProfileScope, ConnectTopCommands, Opt,
|
||||
CommandResult, Config, ConnectClientPerformanceOperation, ConnectClientPerformanceOpts, ConnectDrivePerformanceOpts,
|
||||
ConnectLicenseCommands, ConnectLicenseScopeOpts, ConnectLogsMode, ConnectLogsOpts, ConnectProfileOpts,
|
||||
ConnectProfileTool, ConnectTelemetryArtifactOpts, ConnectTelemetryCommands, ConnectThreadProfileScope,
|
||||
ConnectTopCommands, Opt,
|
||||
},
|
||||
startup_lifecycle::{StartupRuntimeLifecycle, run_startup_runtime_lifecycle},
|
||||
startup_preflight::{StartupServerPreflightError, bootstrap_external_prefix_compat, init_startup_server_preflight},
|
||||
@@ -136,6 +137,7 @@ async fn async_main() -> Result<()> {
|
||||
return Ok(());
|
||||
}
|
||||
CommandResult::ConnectLicense(command) => return execute_connect_license(command),
|
||||
CommandResult::ConnectClientPerformance(options) => return execute_connect_client_performance(options).await,
|
||||
CommandResult::ConnectDrivePerformance(options) => return execute_connect_drive_performance(options).await,
|
||||
CommandResult::ConnectProfile(options) => return execute_connect_profile(options).await,
|
||||
CommandResult::ConnectLogs(options) => return execute_connect_logs(options).await,
|
||||
@@ -606,6 +608,137 @@ async fn finish_top_capture<T: serde::Serialize>(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn execute_connect_client_performance(options: ConnectClientPerformanceOpts) -> Result<()> {
|
||||
use crate::connect::{
|
||||
ClientOperation, ClientOutcome, ClientPerformanceRequest, ClientProvenance, HttpClientProbe, IdentityStore,
|
||||
LocalClientConsent, measure_client, read_protected_client_credential, save_signed_client_export, sign_client_export,
|
||||
validate_client_limits,
|
||||
};
|
||||
use rand::{TryRng as _, rngs::SysRng};
|
||||
use zeroize::Zeroizing;
|
||||
|
||||
let duration = Duration::from_millis(options.duration_millis);
|
||||
validate_client_limits(duration, options.traffic_bytes).map_err(Error::other)?;
|
||||
let key = IdentityStore::new(options.state_dir.join("identity"))
|
||||
.load()
|
||||
.map_err(Error::other)?
|
||||
.ok_or_else(|| Error::other("connect client performance requires an enrolled device identity"))?;
|
||||
let access_key = read_protected_client_credential(&options.access_key_file).map_err(Error::other)?;
|
||||
let secret_key = read_protected_client_credential(&options.secret_key_file).map_err(Error::other)?;
|
||||
let session_token = options
|
||||
.session_token_file
|
||||
.as_deref()
|
||||
.map(read_protected_client_credential)
|
||||
.transpose()
|
||||
.map_err(Error::other)?
|
||||
.unwrap_or_else(|| Zeroizing::new(String::new()));
|
||||
let root_ca = if let Some(path) = options.ca_file.as_deref() {
|
||||
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("connect client root CA exceeds the 1048576-byte limit"));
|
||||
}
|
||||
Some(bytes)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let probe = HttpClientProbe::new(
|
||||
&options.endpoint,
|
||||
root_ca.as_deref(),
|
||||
options.proxy.as_deref(),
|
||||
access_key,
|
||||
secret_key,
|
||||
session_token,
|
||||
duration,
|
||||
)
|
||||
.map_err(Error::other)?;
|
||||
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 = ClientPerformanceRequest {
|
||||
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: LocalClientConsent {
|
||||
consent_uid: options.consent_uid,
|
||||
policy_revision: options.policy_revision,
|
||||
expires_at_unix: options.consent_expires_at_unix,
|
||||
confirmed: options.acknowledge_l1,
|
||||
},
|
||||
produced_at_unix,
|
||||
expires_at_unix: options.expires_at_unix,
|
||||
nonce,
|
||||
duration,
|
||||
operation: match options.operation {
|
||||
ConnectClientPerformanceOperation::Get => ClientOperation::GetObject,
|
||||
ConnectClientPerformanceOperation::Put => ClientOperation::PutObject,
|
||||
},
|
||||
traffic_bytes: options.traffic_bytes,
|
||||
target_alias: options.target_alias,
|
||||
provenance: ClientProvenance::new(
|
||||
crate::version::build::COMMIT_HASH,
|
||||
executable_sha256,
|
||||
env!("CARGO_PKG_VERSION"),
|
||||
enabled_build_features(),
|
||||
),
|
||||
};
|
||||
let cancel = CancellationToken::new();
|
||||
let measurement = measure_client(&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.client outcome={} reason={}",
|
||||
measurement.result.outcome().as_str(),
|
||||
measurement.result.reason_code().as_str()
|
||||
);
|
||||
println!("target={target_json}");
|
||||
std::io::stdout().flush()?;
|
||||
if measurement.result.outcome() != ClientOutcome::Succeeded {
|
||||
return Err(Error::other(format!(
|
||||
"client performance collection ended with {}",
|
||||
measurement.result.outcome().as_str()
|
||||
)));
|
||||
}
|
||||
|
||||
let export = sign_client_export(&request, &measurement, &key, &cancel).map_err(Error::other)?;
|
||||
let output = options.output;
|
||||
let writer_cancel = cancel.clone();
|
||||
let mut writer = tokio::task::spawn_blocking(move || save_signed_client_export(&output, &export, &writer_cancel));
|
||||
let receipt = tokio::select! {
|
||||
biased;
|
||||
signal = tokio::signal::ctrl_c() => {
|
||||
signal.map_err(Error::other)?;
|
||||
cancel.cancel();
|
||||
writer.await.map_err(Error::other)?.map_err(Error::other)?
|
||||
}
|
||||
result = &mut writer => result.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");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn execute_connect_drive_performance(options: ConnectDrivePerformanceOpts) -> Result<()> {
|
||||
use crate::connect::{
|
||||
DriveOutcome, DrivePerformanceRequest, DriveProvenance, IdentityStore, LocalDriveConsent, measure_drive,
|
||||
@@ -808,12 +941,12 @@ async fn execute_connect_profile(options: ConnectProfileOpts) -> Result<()> {
|
||||
fn hash_current_executable() -> Result<String> {
|
||||
use sha2::{Digest as _, Sha256};
|
||||
|
||||
const MAX_EXECUTABLE_BYTES: u64 = 1_073_741_824;
|
||||
const MAX_EXECUTABLE_BYTES: u64 = 2_147_483_648;
|
||||
let path = std::env::current_exe().map_err(Error::other)?;
|
||||
let mut file = std::fs::File::open(path).map_err(Error::other)?;
|
||||
let metadata = file.metadata().map_err(Error::other)?;
|
||||
if !metadata.is_file() || metadata.len() == 0 || metadata.len() > MAX_EXECUTABLE_BYTES {
|
||||
return Err(Error::other("current executable is outside the profile provenance limit"));
|
||||
return Err(Error::other("current executable is outside the diagnostic provenance limit"));
|
||||
}
|
||||
let mut hasher = Sha256::new();
|
||||
let mut buffer = [0_u8; 64 * 1024];
|
||||
@@ -825,14 +958,14 @@ fn hash_current_executable() -> Result<String> {
|
||||
}
|
||||
read_bytes = read_bytes
|
||||
.checked_add(u64::try_from(count).map_err(Error::other)?)
|
||||
.ok_or_else(|| Error::other("current executable is outside the profile provenance limit"))?;
|
||||
.ok_or_else(|| Error::other("current executable is outside the diagnostic provenance limit"))?;
|
||||
if read_bytes > MAX_EXECUTABLE_BYTES {
|
||||
return Err(Error::other("current executable is outside the profile provenance limit"));
|
||||
return Err(Error::other("current executable is outside the diagnostic provenance limit"));
|
||||
}
|
||||
hasher.update(&buffer[..count]);
|
||||
}
|
||||
if read_bytes != metadata.len() {
|
||||
return Err(Error::other("current executable changed while hashing profile provenance"));
|
||||
return Err(Error::other("current executable changed while hashing diagnostic provenance"));
|
||||
}
|
||||
Ok(hex_simd::encode_to_string(hasher.finalize(), hex_simd::AsciiCase::Lower))
|
||||
}
|
||||
|
||||
@@ -0,0 +1,584 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use std::fs::{self, File};
|
||||
use std::io::{Cursor, Read as _};
|
||||
#[cfg(unix)]
|
||||
use std::os::unix::fs::PermissionsExt as _;
|
||||
use std::path::Path;
|
||||
use std::process::Command;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
|
||||
use base64_simd::URL_SAFE_NO_PAD;
|
||||
use p256::ecdsa::{Signature, VerifyingKey, signature::Verifier as _};
|
||||
use p256::pkcs8::DecodePublicKey as _;
|
||||
use rustfs::connect::{
|
||||
CLIENT_CAPABILITY, ClientOperation, ClientOutcome, ClientPerformanceError, ClientPerformanceRequest, ClientProbe,
|
||||
ClientProbeError, ClientProbeFuture, ClientProbeMeasurement, ClientProvenance, ClientReasonCode, ClientTargetReasonCode,
|
||||
DeviceIdentity, HttpClientProbe, LocalClientConsent, measure_client, save_signed_client_export, sign_client_export,
|
||||
validate_client_limits,
|
||||
};
|
||||
use rustfs::embedded::{RustFSServerBuilder, find_available_port};
|
||||
use sha2::{Digest as _, Sha256};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use zeroize::Zeroizing;
|
||||
|
||||
static TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
|
||||
|
||||
fn now() -> i64 {
|
||||
SystemTime::now().duration_since(UNIX_EPOCH).expect("current time").as_secs() as i64
|
||||
}
|
||||
|
||||
fn request(operation: ClientOperation) -> ClientPerformanceRequest {
|
||||
let now = now();
|
||||
let organization = "organizations/019e3ae0-0000-7000-8000-000000000010";
|
||||
let cluster = format!("{organization}/clusters/019e3ae0-0000-7000-8000-000000000011");
|
||||
ClientPerformanceRequest {
|
||||
organization_name: organization.to_owned(),
|
||||
cluster_name: cluster.clone(),
|
||||
device_name: format!("{cluster}/clusterDevices/019e3ae0-0000-7000-8000-000000000012"),
|
||||
run_uid: "019e3ae0-0000-7000-8000-000000000013".to_owned(),
|
||||
artifact_uid: "019e3ae0-0000-7000-8000-000000000014".to_owned(),
|
||||
schema_version: 1,
|
||||
capability: CLIENT_CAPABILITY.to_owned(),
|
||||
consent: LocalClientConsent {
|
||||
consent_uid: "019e3ae0-0000-7000-8000-000000000015".to_owned(),
|
||||
policy_revision: 7,
|
||||
expires_at_unix: now + 120,
|
||||
confirmed: true,
|
||||
},
|
||||
produced_at_unix: now,
|
||||
expires_at_unix: now + 60,
|
||||
nonce: [0x6b; 32],
|
||||
duration: Duration::from_secs(1),
|
||||
operation,
|
||||
traffic_bytes: 65_536,
|
||||
target_alias: "deployment-1".to_owned(),
|
||||
provenance: ClientProvenance::new("c".repeat(40), "d".repeat(64), "1.0.0-rc.6", vec!["default".to_owned()]),
|
||||
}
|
||||
}
|
||||
|
||||
struct SuccessfulProbe;
|
||||
|
||||
impl ClientProbe for SuccessfulProbe {
|
||||
fn probe<'a>(&'a self, request: &'a ClientPerformanceRequest, _cancel: &'a CancellationToken) -> ClientProbeFuture<'a> {
|
||||
Box::pin(async move {
|
||||
Ok(ClientProbeMeasurement {
|
||||
transferred_bytes: request.traffic_bytes,
|
||||
duration: Duration::from_millis(100),
|
||||
latency: Duration::from_millis(7),
|
||||
})
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
struct ErrorProbe(ClientProbeError);
|
||||
|
||||
impl ClientProbe for ErrorProbe {
|
||||
fn probe<'a>(&'a self, _request: &'a ClientPerformanceRequest, _cancel: &'a CancellationToken) -> ClientProbeFuture<'a> {
|
||||
Box::pin(async move { Err(self.0) })
|
||||
}
|
||||
}
|
||||
|
||||
struct PendingProbe;
|
||||
|
||||
impl ClientProbe for PendingProbe {
|
||||
fn probe<'a>(&'a self, _request: &'a ClientPerformanceRequest, _cancel: &'a CancellationToken) -> ClientProbeFuture<'a> {
|
||||
Box::pin(std::future::pending())
|
||||
}
|
||||
}
|
||||
|
||||
struct CountingProbe(AtomicUsize);
|
||||
|
||||
impl ClientProbe for CountingProbe {
|
||||
fn probe<'a>(&'a self, request: &'a ClientPerformanceRequest, _cancel: &'a CancellationToken) -> ClientProbeFuture<'a> {
|
||||
self.0.fetch_add(1, Ordering::Relaxed);
|
||||
Box::pin(async move {
|
||||
Ok(ClientProbeMeasurement {
|
||||
transferred_bytes: request.traffic_bytes,
|
||||
duration: Duration::from_millis(100),
|
||||
latency: Duration::from_millis(1),
|
||||
})
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn typed_get_and_put_results_match_the_frozen_schema() {
|
||||
let _guard = TEST_LOCK.lock().await;
|
||||
for operation in [ClientOperation::GetObject, ClientOperation::PutObject] {
|
||||
let request = request(operation);
|
||||
let measurement = measure_client(&request, &SuccessfulProbe, &CancellationToken::new())
|
||||
.await
|
||||
.expect("client measurement");
|
||||
assert_eq!(measurement.result.outcome(), ClientOutcome::Succeeded);
|
||||
assert_eq!(measurement.result.reason_code(), ClientReasonCode::Complete);
|
||||
let data = measurement.result.data().expect("aggregate data");
|
||||
assert_eq!(data.operation, operation);
|
||||
assert_eq!(data.transferred_bytes, 65_536);
|
||||
assert_eq!(data.completed_operations, 1);
|
||||
assert_eq!(data.duration_millis, 100);
|
||||
assert_eq!(data.error_count, 0);
|
||||
assert_eq!(measurement.target.target_alias, "deployment-1");
|
||||
assert_eq!(measurement.target.parameters.operation, operation);
|
||||
assert_eq!(measurement.target.parameters.requested_bytes, 65_536);
|
||||
assert_eq!(measurement.target.parameters.concurrency, 1);
|
||||
assert_eq!(measurement.target.latency_micros, Some(7_000));
|
||||
|
||||
let value = serde_json::to_value(&measurement.result).expect("result JSON");
|
||||
assert_eq!(value["toolId"], "performance.client");
|
||||
assert_eq!(value["capability"], "performance.client@1");
|
||||
assert_eq!(value["coverage"]["unit"], "WINDOW");
|
||||
assert_eq!(value["data"]["operation"], operation.as_str());
|
||||
assert!(value["data"].get("latencyMicros").is_none());
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn consent_and_budget_fail_before_transport() {
|
||||
let _guard = TEST_LOCK.lock().await;
|
||||
let probe = CountingProbe(AtomicUsize::new(0));
|
||||
|
||||
let mut unsupported_version = request(ClientOperation::PutObject);
|
||||
unsupported_version.schema_version = 2;
|
||||
assert!(matches!(
|
||||
measure_client(&unsupported_version, &probe, &CancellationToken::new()).await,
|
||||
Err(ClientPerformanceError::UnsupportedVersion)
|
||||
));
|
||||
|
||||
let mut unsupported_capability = request(ClientOperation::PutObject);
|
||||
unsupported_capability.capability = "performance.client@2".to_owned();
|
||||
assert!(matches!(
|
||||
measure_client(&unsupported_capability, &probe, &CancellationToken::new()).await,
|
||||
Err(ClientPerformanceError::UnsupportedCapability)
|
||||
));
|
||||
|
||||
let mut no_consent = request(ClientOperation::PutObject);
|
||||
no_consent.consent.confirmed = false;
|
||||
assert!(matches!(
|
||||
measure_client(&no_consent, &probe, &CancellationToken::new()).await,
|
||||
Err(ClientPerformanceError::ConsentRequired)
|
||||
));
|
||||
|
||||
let mut over_traffic = request(ClientOperation::PutObject);
|
||||
over_traffic.traffic_bytes = 1_048_577;
|
||||
assert!(matches!(
|
||||
measure_client(&over_traffic, &probe, &CancellationToken::new()).await,
|
||||
Err(ClientPerformanceError::LimitExceeded)
|
||||
));
|
||||
|
||||
let mut invalid_version = request(ClientOperation::PutObject);
|
||||
invalid_version.provenance =
|
||||
ClientProvenance::new("a".repeat(40), "b".repeat(64), "release_candidate", vec!["default".to_owned()]);
|
||||
assert!(matches!(
|
||||
measure_client(&invalid_version, &probe, &CancellationToken::new()).await,
|
||||
Err(ClientPerformanceError::InvalidRequest)
|
||||
));
|
||||
|
||||
let mut invalid_feature = request(ClientOperation::PutObject);
|
||||
invalid_feature.provenance = ClientProvenance::new("a".repeat(40), "b".repeat(64), "1.0.0-rc.6", vec!["Default".to_owned()]);
|
||||
assert!(matches!(
|
||||
measure_client(&invalid_feature, &probe, &CancellationToken::new()).await,
|
||||
Err(ClientPerformanceError::InvalidRequest)
|
||||
));
|
||||
assert_eq!(probe.0.load(Ordering::Relaxed), 0);
|
||||
|
||||
assert!(validate_client_limits(Duration::from_secs(1), 1_048_576).is_ok());
|
||||
assert!(matches!(
|
||||
validate_client_limits(Duration::from_secs(1), 1_048_577),
|
||||
Err(ClientPerformanceError::LimitExceeded)
|
||||
));
|
||||
assert!(matches!(
|
||||
validate_client_limits(Duration::from_nanos(1), 1),
|
||||
Err(ClientPerformanceError::LimitExceeded)
|
||||
));
|
||||
assert!(matches!(
|
||||
validate_client_limits(Duration::from_secs(30) + Duration::from_nanos(1), 1),
|
||||
Err(ClientPerformanceError::LimitExceeded)
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn endpoint_proxy_permission_timeout_and_cancel_are_explicit() {
|
||||
let _guard = TEST_LOCK.lock().await;
|
||||
for (error, expected_reason, expected_result_reason) in [
|
||||
(
|
||||
ClientProbeError::EndpointUnavailable,
|
||||
ClientTargetReasonCode::EndpointUnavailable,
|
||||
ClientReasonCode::SourceUnavailable,
|
||||
),
|
||||
(
|
||||
ClientProbeError::ProxyFailure,
|
||||
ClientTargetReasonCode::ProxyFailure,
|
||||
ClientReasonCode::CollectionFailed,
|
||||
),
|
||||
(
|
||||
ClientProbeError::PermissionDenied,
|
||||
ClientTargetReasonCode::PermissionDenied,
|
||||
ClientReasonCode::PermissionDenied,
|
||||
),
|
||||
(
|
||||
ClientProbeError::TimedOut,
|
||||
ClientTargetReasonCode::TimedOut,
|
||||
ClientReasonCode::CollectionFailed,
|
||||
),
|
||||
] {
|
||||
let measurement = measure_client(&request(ClientOperation::PutObject), &ErrorProbe(error), &CancellationToken::new())
|
||||
.await
|
||||
.expect("typed failure");
|
||||
assert_eq!(measurement.result.outcome(), ClientOutcome::Failed);
|
||||
assert_eq!(measurement.result.reason_code(), expected_result_reason);
|
||||
assert_eq!(measurement.target.reason_code, expected_reason);
|
||||
assert!(measurement.result.data().is_none());
|
||||
}
|
||||
|
||||
let cancel = CancellationToken::new();
|
||||
cancel.cancel();
|
||||
let measurement = measure_client(&request(ClientOperation::PutObject), &SuccessfulProbe, &cancel)
|
||||
.await
|
||||
.expect("typed cancellation");
|
||||
assert_eq!(measurement.result.outcome(), ClientOutcome::Cancelled);
|
||||
assert_eq!(measurement.target.reason_code, ClientTargetReasonCode::Cancelled);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn deadline_and_in_flight_cancellation_stop_a_stalled_probe() {
|
||||
let _guard = TEST_LOCK.lock().await;
|
||||
let mut timed = request(ClientOperation::GetObject);
|
||||
timed.duration = Duration::from_millis(20);
|
||||
timed.traffic_bytes = 1_024;
|
||||
let measurement = measure_client(&timed, &PendingProbe, &CancellationToken::new())
|
||||
.await
|
||||
.expect("typed timeout");
|
||||
assert_eq!(measurement.result.outcome(), ClientOutcome::Failed);
|
||||
assert_eq!(measurement.target.reason_code, ClientTargetReasonCode::TimedOut);
|
||||
|
||||
let cancel = CancellationToken::new();
|
||||
let cancellation = cancel.clone();
|
||||
tokio::spawn(async move {
|
||||
tokio::task::yield_now().await;
|
||||
cancellation.cancel();
|
||||
});
|
||||
let measurement = measure_client(&request(ClientOperation::PutObject), &PendingProbe, &cancel)
|
||||
.await
|
||||
.expect("typed in-flight cancellation");
|
||||
assert_eq!(measurement.result.outcome(), ClientOutcome::Cancelled);
|
||||
assert_eq!(measurement.target.reason_code, ClientTargetReasonCode::Cancelled);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn only_one_client_collector_can_run_at_a_time() {
|
||||
let _guard = TEST_LOCK.lock().await;
|
||||
let first_cancel = CancellationToken::new();
|
||||
let second_cancel = CancellationToken::new();
|
||||
let first_request = request(ClientOperation::GetObject);
|
||||
let second_request = request(ClientOperation::PutObject);
|
||||
let first = measure_client(&first_request, &PendingProbe, &first_cancel);
|
||||
let second = async {
|
||||
tokio::task::yield_now().await;
|
||||
measure_client(&second_request, &PendingProbe, &second_cancel).await
|
||||
};
|
||||
let cancellation = async {
|
||||
tokio::time::sleep(Duration::from_millis(20)).await;
|
||||
first_cancel.cancel();
|
||||
};
|
||||
let (first, second, ()) = tokio::join!(first, second, cancellation);
|
||||
assert_eq!(first.expect("typed cancellation").result.outcome(), ClientOutcome::Cancelled);
|
||||
assert!(matches!(second, Err(ClientPerformanceError::Busy)));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn configured_proxy_and_direct_endpoint_failures_are_distinct() {
|
||||
let _guard = TEST_LOCK.lock().await;
|
||||
let proxy_probe = HttpClientProbe::new(
|
||||
"http://127.0.0.1:1",
|
||||
None,
|
||||
Some("http://127.0.0.1:9"),
|
||||
Zeroizing::new("access".to_owned()),
|
||||
Zeroizing::new("secret".to_owned()),
|
||||
Zeroizing::new(String::new()),
|
||||
Duration::from_millis(100),
|
||||
)
|
||||
.expect("proxy probe");
|
||||
let measurement = measure_client(&request(ClientOperation::GetObject), &proxy_probe, &CancellationToken::new())
|
||||
.await
|
||||
.expect("typed proxy failure");
|
||||
assert_eq!(measurement.target.reason_code, ClientTargetReasonCode::ProxyFailure);
|
||||
|
||||
let direct_probe = HttpClientProbe::new(
|
||||
"http://127.0.0.1:9",
|
||||
None,
|
||||
None,
|
||||
Zeroizing::new("access".to_owned()),
|
||||
Zeroizing::new("secret".to_owned()),
|
||||
Zeroizing::new(String::new()),
|
||||
Duration::from_millis(100),
|
||||
)
|
||||
.expect("direct probe");
|
||||
let measurement = measure_client(&request(ClientOperation::GetObject), &direct_probe, &CancellationToken::new())
|
||||
.await
|
||||
.expect("typed endpoint failure");
|
||||
assert_eq!(measurement.target.reason_code, ClientTargetReasonCode::EndpointUnavailable);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn oversized_response_is_rejected_without_buffering_it() {
|
||||
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
|
||||
|
||||
let _guard = TEST_LOCK.lock().await;
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("response listener");
|
||||
let address = listener.local_addr().expect("listener address");
|
||||
let server = tokio::spawn(async move {
|
||||
let (mut socket, _) = listener.accept().await.expect("client connection");
|
||||
let mut request = vec![0_u8; 16 * 1024];
|
||||
let _ = socket.read(&mut request).await.expect("request headers");
|
||||
socket
|
||||
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 65537\r\nConnection: close\r\n\r\n")
|
||||
.await
|
||||
.expect("response headers");
|
||||
});
|
||||
let probe = HttpClientProbe::new(
|
||||
&format!("http://{address}"),
|
||||
None,
|
||||
None,
|
||||
Zeroizing::new("access".to_owned()),
|
||||
Zeroizing::new("secret".to_owned()),
|
||||
Zeroizing::new(String::new()),
|
||||
Duration::from_secs(1),
|
||||
)
|
||||
.expect("client probe");
|
||||
let measurement = measure_client(&request(ClientOperation::GetObject), &probe, &CancellationToken::new())
|
||||
.await
|
||||
.expect("typed protocol failure");
|
||||
assert_eq!(measurement.target.reason_code, ClientTargetReasonCode::ProtocolFailure);
|
||||
server.await.expect("response server");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn signed_result_is_saved_without_overwrite() {
|
||||
let _guard = TEST_LOCK.lock().await;
|
||||
let request = request(ClientOperation::PutObject);
|
||||
let measurement = measure_client(&request, &SuccessfulProbe, &CancellationToken::new())
|
||||
.await
|
||||
.expect("client measurement");
|
||||
let export = sign_client_export(&request, &measurement, &DeviceIdentity::generate(), &CancellationToken::new())
|
||||
.expect("signed export");
|
||||
let directory = tempfile::tempdir().expect("output directory");
|
||||
let output = directory.path().join("client.zip");
|
||||
let saved = save_signed_client_export(&output, &export, &CancellationToken::new()).expect("save export");
|
||||
assert_eq!(saved.archive_sha256, export.archive_sha256);
|
||||
assert!(matches!(
|
||||
save_signed_client_export(&output, &export, &CancellationToken::new()),
|
||||
Err(ClientPerformanceError::AlreadyExists)
|
||||
));
|
||||
|
||||
let mut tampered = export.clone();
|
||||
tampered.archive_bytes[0] ^= 0xff;
|
||||
assert!(matches!(
|
||||
save_signed_client_export(&directory.path().join("tampered.zip"), &tampered, &CancellationToken::new()),
|
||||
Err(ClientPerformanceError::InvalidRequest)
|
||||
));
|
||||
|
||||
let mut oversized = export.clone();
|
||||
oversized.archive_bytes = vec![0; 524_289];
|
||||
oversized.archive_sha256 = hex_lower(&Sha256::digest(&oversized.archive_bytes));
|
||||
assert!(matches!(
|
||||
save_signed_client_export(&directory.path().join("oversized.zip"), &oversized, &CancellationToken::new()),
|
||||
Err(ClientPerformanceError::LimitExceeded)
|
||||
));
|
||||
|
||||
let cancelled = CancellationToken::new();
|
||||
cancelled.cancel();
|
||||
assert!(matches!(
|
||||
save_signed_client_export(&directory.path().join("cancelled.zip"), &export, &cancelled),
|
||||
Err(ClientPerformanceError::Cancelled)
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn real_rustfs_endpoint_supports_bounded_get_and_put() {
|
||||
let _guard = TEST_LOCK.lock().await;
|
||||
let port = match find_available_port() {
|
||||
Ok(port) => port,
|
||||
Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return,
|
||||
Err(err) => panic!("find free port: {err}"),
|
||||
};
|
||||
let server = RustFSServerBuilder::new()
|
||||
.address(format!("127.0.0.1:{port}"))
|
||||
.access_key("client-perf-access")
|
||||
.secret_key("client-perf-secret")
|
||||
.build()
|
||||
.await
|
||||
.expect("start embedded server");
|
||||
let probe = HttpClientProbe::new(
|
||||
&server.endpoint(),
|
||||
None,
|
||||
None,
|
||||
Zeroizing::new(server.access_key().to_owned()),
|
||||
Zeroizing::new(server.secret_key().to_owned()),
|
||||
Zeroizing::new(String::new()),
|
||||
Duration::from_secs(2),
|
||||
)
|
||||
.expect("client probe");
|
||||
|
||||
for operation in [ClientOperation::GetObject, ClientOperation::PutObject] {
|
||||
let mut request = request(operation);
|
||||
request.duration = Duration::from_secs(2);
|
||||
let measurement = measure_client(&request, &probe, &CancellationToken::new())
|
||||
.await
|
||||
.expect("real client measurement");
|
||||
assert_eq!(measurement.result.outcome(), ClientOutcome::Succeeded);
|
||||
assert_eq!(measurement.result.data().expect("data").transferred_bytes, 65_536);
|
||||
assert_eq!(measurement.target.completed_operations, 1);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn production_cli_writes_a_verifiable_export_with_exact_binary_provenance() {
|
||||
let _guard = TEST_LOCK.lock().await;
|
||||
let port = match find_available_port() {
|
||||
Ok(port) => port,
|
||||
Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return,
|
||||
Err(err) => panic!("find free port: {err}"),
|
||||
};
|
||||
let server = RustFSServerBuilder::new()
|
||||
.address(format!("127.0.0.1:{port}"))
|
||||
.access_key("client-perf-access")
|
||||
.secret_key("client-perf-secret")
|
||||
.build()
|
||||
.await
|
||||
.expect("start embedded server");
|
||||
let temp = tempfile::tempdir().expect("CLI tempdir");
|
||||
let state = temp.path().join("state");
|
||||
let output = temp.path().join("client.zip");
|
||||
let access_key_file = temp.path().join("access-key");
|
||||
let secret_key_file = temp.path().join("secret-key");
|
||||
write_credential(&access_key_file, server.access_key());
|
||||
write_credential(&secret_key_file, server.secret_key());
|
||||
let identity = rustfs::connect::IdentityStore::new(state.join("identity"))
|
||||
.load_or_create()
|
||||
.expect("enrolled identity");
|
||||
|
||||
let current = now();
|
||||
let organization = "organizations/019e3ae0-0000-7000-8000-000000000010";
|
||||
let cluster = format!("{organization}/clusters/019e3ae0-0000-7000-8000-000000000011");
|
||||
let mut command = Command::new(env!("CARGO_BIN_EXE_rustfs"));
|
||||
command
|
||||
.args(["connect", "performance", "client", "--state-dir"])
|
||||
.arg(&state)
|
||||
.args(["--endpoint", &server.endpoint(), "--access-key-file"])
|
||||
.arg(&access_key_file)
|
||||
.arg("--secret-key-file")
|
||||
.arg(&secret_key_file)
|
||||
.arg("--output")
|
||||
.arg(&output)
|
||||
.args(["--organization", organization, "--cluster", &cluster, "--device"])
|
||||
.arg(format!("{cluster}/clusterDevices/019e3ae0-0000-7000-8000-000000000012"))
|
||||
.args([
|
||||
"--run-uid",
|
||||
"019e3ae0-0000-7000-8000-000000000013",
|
||||
"--artifact-uid",
|
||||
"019e3ae0-0000-7000-8000-000000000014",
|
||||
"--consent-uid",
|
||||
"019e3ae0-0000-7000-8000-000000000015",
|
||||
"--policy-revision",
|
||||
"7",
|
||||
"--consent-expires-at",
|
||||
&(current + 120).to_string(),
|
||||
"--expires-at",
|
||||
&(current + 60).to_string(),
|
||||
"--operation",
|
||||
"get",
|
||||
"--traffic-bytes",
|
||||
"65536",
|
||||
"--duration-millis",
|
||||
"1000",
|
||||
"--acknowledge-l1",
|
||||
]);
|
||||
let result = tokio::task::spawn_blocking(move || command.output())
|
||||
.await
|
||||
.expect("CLI task")
|
||||
.expect("run production rustfs binary");
|
||||
|
||||
assert!(result.status.success(), "stderr: {}", String::from_utf8_lossy(&result.stderr));
|
||||
let stdout = String::from_utf8(result.stdout).expect("UTF-8 stdout");
|
||||
assert!(stdout.contains("tool=performance.client outcome=SUCCEEDED reason=COMPLETE\n"));
|
||||
assert!(stdout.contains("upload=not-performed\n"));
|
||||
#[cfg(unix)]
|
||||
assert_eq!(fs::metadata(&output).expect("output metadata").permissions().mode(), 0o100600);
|
||||
|
||||
let archive_bytes = fs::read(&output).expect("saved archive");
|
||||
let mut archive = zip::ZipArchive::new(Cursor::new(archive_bytes.as_slice())).expect("signed archive");
|
||||
let envelope_bytes = read_archive_member(&mut archive, "envelope.json");
|
||||
let signature_bytes = read_archive_member(&mut archive, "envelope.sig");
|
||||
let result_bytes = read_archive_member(&mut archive, "result.json");
|
||||
let envelope: serde_json::Value = serde_json::from_slice(&envelope_bytes).expect("envelope JSON");
|
||||
let signed_result: serde_json::Value = serde_json::from_slice(&result_bytes).expect("result JSON");
|
||||
assert_eq!(envelope["classification"], "L1");
|
||||
assert_eq!(envelope["payload"]["sha256"], hex_lower(&Sha256::digest(&result_bytes)));
|
||||
assert_eq!(signed_result["data"]["operation"], "GET_OBJECT");
|
||||
assert_eq!(signed_result["data"]["transferredBytes"], 65_536);
|
||||
assert_eq!(signed_result["provenance"]["sourceCommit"], rustfs::version::build::COMMIT_HASH);
|
||||
assert_eq!(
|
||||
signed_result["provenance"]["executableSha256"],
|
||||
sha256_file(Path::new(env!("CARGO_BIN_EXE_rustfs")))
|
||||
);
|
||||
|
||||
let signature_document: serde_json::Value = serde_json::from_slice(&signature_bytes).expect("signature JSON");
|
||||
let raw = URL_SAFE_NO_PAD
|
||||
.decode_to_vec(signature_document["value"].as_str().expect("signature value"))
|
||||
.expect("base64url signature");
|
||||
let signature = Signature::from_slice(&raw).expect("P-256 signature");
|
||||
let mut signed = b"rustfs-diagnostic-envelope-v1\0".to_vec();
|
||||
signed.extend_from_slice(&envelope_bytes);
|
||||
VerifyingKey::from_public_key_der(&identity.public_key_der())
|
||||
.expect("public key")
|
||||
.verify(&signed, &signature)
|
||||
.expect("valid ES256 signature");
|
||||
}
|
||||
|
||||
fn write_credential(path: &Path, value: &str) {
|
||||
fs::write(path, value).expect("write credential");
|
||||
#[cfg(unix)]
|
||||
fs::set_permissions(path, fs::Permissions::from_mode(0o600)).expect("protect credential");
|
||||
}
|
||||
|
||||
fn read_archive_member(archive: &mut zip::ZipArchive<Cursor<&[u8]>>, name: &str) -> Vec<u8> {
|
||||
let mut bytes = Vec::new();
|
||||
archive
|
||||
.by_name(name)
|
||||
.expect("archive member")
|
||||
.read_to_end(&mut bytes)
|
||||
.expect("read archive member");
|
||||
bytes
|
||||
}
|
||||
|
||||
fn sha256_file(path: &Path) -> String {
|
||||
let mut file = File::open(path).expect("open exact binary");
|
||||
let mut digest = Sha256::new();
|
||||
let mut buffer = [0_u8; 64 * 1024];
|
||||
loop {
|
||||
let read = file.read(&mut buffer).expect("hash exact binary");
|
||||
if read == 0 {
|
||||
break;
|
||||
}
|
||||
digest.update(&buffer[..read]);
|
||||
}
|
||||
hex_lower(&digest.finalize())
|
||||
}
|
||||
|
||||
fn hex_lower(bytes: &[u8]) -> String {
|
||||
hex_simd::encode_to_string(bytes, hex_simd::AsciiCase::Lower)
|
||||
}
|
||||
Reference in New Issue
Block a user