From 8ae8de0d281d43f9e00734b5f29fa6fe2c239f79 Mon Sep 17 00:00:00 2001 From: Chris Date: Sun, 13 Sep 2026 13:17:15 +0800 Subject: [PATCH] feat(connect): add bounded object performance diagnostics (#7734) feat(connect): add object performance producer --- protocol/agent/v1/fixtures/fixture-sets.json | 5 + protocol/agent/v1/fixtures/manifest.sha256 | 9 + .../object-performance/MANIFEST.sha256 | 2 + .../object-performance/accept-vectors.json | 222 +++ .../object-performance/reject-vectors.json | 319 ++++ rustfs/src/config/cli.rs | 109 ++ rustfs/src/config/mod.rs | 1 + rustfs/src/config/opt.rs | 1 + rustfs/src/connect/diagnostics/mod.rs | 10 + rustfs/src/connect/diagnostics/perf_object.rs | 1345 +++++++++++++++++ rustfs/src/connect/mod.rs | 9 + rustfs/src/startup_entrypoint.rs | 137 +- rustfs/tests/agent_protocol_fixtures.rs | 7 +- rustfs/tests/connect_perf_object.rs | 628 ++++++++ 14 files changed, 2799 insertions(+), 5 deletions(-) create mode 100644 protocol/agent/v1/fixtures/manifest.sha256 create mode 100644 protocol/agent/v1/fixtures/object-performance/MANIFEST.sha256 create mode 100644 protocol/agent/v1/fixtures/object-performance/accept-vectors.json create mode 100644 protocol/agent/v1/fixtures/object-performance/reject-vectors.json create mode 100644 rustfs/src/connect/diagnostics/perf_object.rs create mode 100644 rustfs/tests/connect_perf_object.rs diff --git a/protocol/agent/v1/fixtures/fixture-sets.json b/protocol/agent/v1/fixtures/fixture-sets.json index 72c4c1919..844076def 100644 --- a/protocol/agent/v1/fixtures/fixture-sets.json +++ b/protocol/agent/v1/fixtures/fixture-sets.json @@ -33,6 +33,11 @@ "status": "populated", "purpose": "Inventory snapshot payloads and their allow-listed collection fields." }, + { + "name": "object-performance", + "status": "populated", + "purpose": "Bounded performance.object@1 result acceptance and rejection vectors." + }, { "name": "offline-enrollment", "status": "populated", diff --git a/protocol/agent/v1/fixtures/manifest.sha256 b/protocol/agent/v1/fixtures/manifest.sha256 new file mode 100644 index 000000000..aae83a1ea --- /dev/null +++ b/protocol/agent/v1/fixtures/manifest.sha256 @@ -0,0 +1,9 @@ +a499742c06046e9cf2f781fc17d8b4cf95e018ab45fa41ee5944ba15b0e2902c auth +9b54e39584a442270ae486e8378df1d6f168c45d0c59baba9459b46bb212d2df version +159d116e965b51771f1687fc6ec9bdfbf2a4a0e6d6aa26d3f1046b3dac2697dd registration +f707b09b257c7d652eff2a3f680ecb0c8c45ebb4c2defa8dc07f9dd7ee46ca04 heartbeat +cc6c3f3bf5e6a938f2e00b8ce42d8fa7bfb974f4eb2927101d4ede4ebdc10dd5 inventory +0207458ec9b97b368e2f347da112511a3aabd27d752890cacda4fa2408e7f19e object-performance +a2557ab1f7f70fb86a8affc8451596c272478fd2d5500c367b8f9e8385d72578 offline-enrollment +04303e0b3c0d1a9de6957555c9ddd6d96b98be2b009de79ce223dd56b81e1909 bundle +42cc936dc8fe87335ebff5abe01e3ebf85bdada7f37b43bfd8add46435c9d6f9 redaction diff --git a/protocol/agent/v1/fixtures/object-performance/MANIFEST.sha256 b/protocol/agent/v1/fixtures/object-performance/MANIFEST.sha256 new file mode 100644 index 000000000..d1768292b --- /dev/null +++ b/protocol/agent/v1/fixtures/object-performance/MANIFEST.sha256 @@ -0,0 +1,2 @@ +24cd96339f27034a0e2bb656ee0c7da9bec9326ded4583a3c9afef06595626b2 accept-vectors.json +a9f4376ea5bbf636851e683cdfaca0cb7ebbc9dfce8ffa4dc8b4a12f82307a53 reject-vectors.json diff --git a/protocol/agent/v1/fixtures/object-performance/accept-vectors.json b/protocol/agent/v1/fixtures/object-performance/accept-vectors.json new file mode 100644 index 000000000..20cc02ccc --- /dev/null +++ b/protocol/agent/v1/fixtures/object-performance/accept-vectors.json @@ -0,0 +1,222 @@ +{ + "protocolVersion": "v1", + "fixtureSet": "object-performance", + "toolId": "performance.object", + "schemaVersion": 1, + "capability": "performance.object@1", + "classification": "L1", + "description": "Frozen performance.object@1 result vectors. Payload bytes are synthetic and remain on the customer deployment; bucket, object, endpoint, credential, and free-text fields are forbidden.", + "fixture": "accept-vectors", + "vectors": [ + { + "id": "performance.object.succeeded", + "schemaValid": true, + "value": { + "schemaVersion": 1, + "runUid": "019e3ae0-0000-7000-8000-000000000001", + "toolId": "performance.object", + "capability": "performance.object@1", + "outcome": "SUCCEEDED", + "reasonCode": "COMPLETE", + "durationMillis": 1000, + "provenance": { + "repository": "rustfs/rustfs", + "sourceCommit": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "executableSha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "rustfsVersion": "1.0.0-rc.6", + "osFamily": "LINUX", + "architecture": "aarch64", + "buildFeatures": [] + }, + "coverage": { + "requestedUnits": 1, + "completedUnits": 1, + "unit": "WINDOW" + }, + "data": { + "operation": "GET_OBJECT", + "transferredBytes": 1048576, + "completedOperations": 1, + "durationMillis": 1000, + "errorCount": 0 + } + }, + "expected": { + "accepted": true, + "reason": null + } + }, + { + "id": "performance.object.put-succeeded", + "schemaValid": true, + "value": { + "schemaVersion": 1, + "runUid": "019e3ae0-0000-7000-8000-000000000001", + "toolId": "performance.object", + "capability": "performance.object@1", + "outcome": "SUCCEEDED", + "reasonCode": "COMPLETE", + "durationMillis": 1000, + "provenance": { + "repository": "rustfs/rustfs", + "sourceCommit": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "executableSha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "rustfsVersion": "1.0.0-rc.6", + "osFamily": "LINUX", + "architecture": "aarch64", + "buildFeatures": [] + }, + "coverage": { + "requestedUnits": 1, + "completedUnits": 1, + "unit": "WINDOW" + }, + "data": { + "operation": "PUT_OBJECT", + "transferredBytes": 1048576, + "completedOperations": 1, + "durationMillis": 1000, + "errorCount": 0 + } + }, + "expected": { + "accepted": true, + "reason": null + } + }, + { + "id": "performance.object.partial", + "schemaValid": true, + "value": { + "schemaVersion": 1, + "runUid": "019e3ae0-0000-7000-8000-000000000001", + "toolId": "performance.object", + "capability": "performance.object@1", + "outcome": "PARTIAL", + "reasonCode": "LIMIT_EXCEEDED", + "durationMillis": 1000, + "provenance": { + "repository": "rustfs/rustfs", + "sourceCommit": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "executableSha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "rustfsVersion": "1.0.0-rc.6", + "osFamily": "LINUX", + "architecture": "aarch64", + "buildFeatures": [] + }, + "coverage": { + "requestedUnits": 2, + "completedUnits": 1, + "unit": "WINDOW" + }, + "data": { + "operation": "GET_OBJECT", + "transferredBytes": 1048576, + "completedOperations": 1, + "durationMillis": 1000, + "errorCount": 0 + } + }, + "expected": { + "accepted": true, + "reason": null + } + }, + { + "id": "performance.object.failed", + "schemaValid": true, + "value": { + "schemaVersion": 1, + "runUid": "019e3ae0-0000-7000-8000-000000000001", + "toolId": "performance.object", + "capability": "performance.object@1", + "outcome": "FAILED", + "reasonCode": "COLLECTION_FAILED", + "durationMillis": 1000, + "provenance": { + "repository": "rustfs/rustfs", + "sourceCommit": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "executableSha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "rustfsVersion": "1.0.0-rc.6", + "osFamily": "LINUX", + "architecture": "aarch64", + "buildFeatures": [] + }, + "coverage": { + "requestedUnits": 1, + "completedUnits": 0, + "unit": "WINDOW" + }, + "data": null + }, + "expected": { + "accepted": true, + "reason": null + } + }, + { + "id": "performance.object.unsupported", + "schemaValid": true, + "value": { + "schemaVersion": 1, + "runUid": "019e3ae0-0000-7000-8000-000000000001", + "toolId": "performance.object", + "capability": "performance.object@1", + "outcome": "UNSUPPORTED", + "reasonCode": "UNSUPPORTED_TOOL", + "durationMillis": 1000, + "provenance": { + "repository": "rustfs/rustfs", + "sourceCommit": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "executableSha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "rustfsVersion": "1.0.0-rc.6", + "osFamily": "LINUX", + "architecture": "aarch64", + "buildFeatures": [] + }, + "coverage": { + "requestedUnits": 1, + "completedUnits": 0, + "unit": "WINDOW" + }, + "data": null + }, + "expected": { + "accepted": true, + "reason": null + } + }, + { + "id": "performance.object.cancelled", + "schemaValid": true, + "value": { + "schemaVersion": 1, + "runUid": "019e3ae0-0000-7000-8000-000000000001", + "toolId": "performance.object", + "capability": "performance.object@1", + "outcome": "CANCELLED", + "reasonCode": "CANCELLED", + "durationMillis": 1000, + "provenance": { + "repository": "rustfs/rustfs", + "sourceCommit": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "executableSha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "rustfsVersion": "1.0.0-rc.6", + "osFamily": "LINUX", + "architecture": "aarch64", + "buildFeatures": [] + }, + "coverage": { + "requestedUnits": 1, + "completedUnits": 0, + "unit": "WINDOW" + }, + "data": null + }, + "expected": { + "accepted": true, + "reason": null + } + } + ] +} diff --git a/protocol/agent/v1/fixtures/object-performance/reject-vectors.json b/protocol/agent/v1/fixtures/object-performance/reject-vectors.json new file mode 100644 index 000000000..da77f847b --- /dev/null +++ b/protocol/agent/v1/fixtures/object-performance/reject-vectors.json @@ -0,0 +1,319 @@ +{ + "protocolVersion": "v1", + "fixtureSet": "object-performance", + "toolId": "performance.object", + "schemaVersion": 1, + "capability": "performance.object@1", + "classification": "L1", + "description": "Frozen performance.object@1 result vectors. Payload bytes are synthetic and remain on the customer deployment; bucket, object, endpoint, credential, and free-text fields are forbidden.", + "fixture": "reject-vectors", + "vectors": [ + { + "id": "performance.object.unknown-data-field", + "schemaValid": false, + "value": { + "schemaVersion": 1, + "runUid": "019e3ae0-0000-7000-8000-000000000001", + "toolId": "performance.object", + "capability": "performance.object@1", + "outcome": "SUCCEEDED", + "reasonCode": "COMPLETE", + "durationMillis": 1000, + "provenance": { + "repository": "rustfs/rustfs", + "sourceCommit": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "executableSha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "rustfsVersion": "1.0.0-rc.6", + "osFamily": "LINUX", + "architecture": "aarch64", + "buildFeatures": [] + }, + "coverage": { + "requestedUnits": 1, + "completedUnits": 1, + "unit": "WINDOW" + }, + "data": { + "operation": "GET_OBJECT", + "transferredBytes": 1048576, + "completedOperations": 1, + "durationMillis": 1000, + "errorCount": 0, + "rawData": "SYNTHETIC_FORBIDDEN_BYTES" + } + }, + "expected": { + "accepted": false, + "reason": "UNKNOWN_FIELD" + } + }, + { + "id": "performance.object.unknown-version", + "schemaValid": false, + "value": { + "schemaVersion": 2, + "runUid": "019e3ae0-0000-7000-8000-000000000001", + "toolId": "performance.object", + "capability": "performance.object@1", + "outcome": "SUCCEEDED", + "reasonCode": "COMPLETE", + "durationMillis": 1000, + "provenance": { + "repository": "rustfs/rustfs", + "sourceCommit": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "executableSha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "rustfsVersion": "1.0.0-rc.6", + "osFamily": "LINUX", + "architecture": "aarch64", + "buildFeatures": [] + }, + "coverage": { + "requestedUnits": 1, + "completedUnits": 1, + "unit": "WINDOW" + }, + "data": { + "operation": "GET_OBJECT", + "transferredBytes": 1048576, + "completedOperations": 1, + "durationMillis": 1000, + "errorCount": 0 + } + }, + "expected": { + "accepted": false, + "reason": "UNSUPPORTED_SCHEMA" + } + }, + { + "id": "performance.object.unsupported-with-data", + "schemaValid": false, + "value": { + "schemaVersion": 1, + "runUid": "019e3ae0-0000-7000-8000-000000000001", + "toolId": "performance.object", + "capability": "performance.object@1", + "outcome": "UNSUPPORTED", + "reasonCode": "UNSUPPORTED_TOOL", + "durationMillis": 1000, + "provenance": { + "repository": "rustfs/rustfs", + "sourceCommit": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "executableSha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "rustfsVersion": "1.0.0-rc.6", + "osFamily": "LINUX", + "architecture": "aarch64", + "buildFeatures": [] + }, + "coverage": { + "requestedUnits": 1, + "completedUnits": 1, + "unit": "WINDOW" + }, + "data": { + "operation": "GET_OBJECT", + "transferredBytes": 1048576, + "completedOperations": 1, + "durationMillis": 1000, + "errorCount": 0 + } + }, + "expected": { + "accepted": false, + "reason": "INVALID_OUTCOME_DATA" + } + }, + { + "id": "performance.object.error-count-exceeds-operations", + "schemaValid": true, + "value": { + "schemaVersion": 1, + "runUid": "019e3ae0-0000-7000-8000-000000000001", + "toolId": "performance.object", + "capability": "performance.object@1", + "outcome": "SUCCEEDED", + "reasonCode": "COMPLETE", + "durationMillis": 1000, + "provenance": { + "repository": "rustfs/rustfs", + "sourceCommit": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "executableSha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "rustfsVersion": "1.0.0-rc.6", + "osFamily": "LINUX", + "architecture": "aarch64", + "buildFeatures": [] + }, + "coverage": { + "requestedUnits": 1, + "completedUnits": 1, + "unit": "WINDOW" + }, + "data": { + "operation": "GET_OBJECT", + "transferredBytes": 1048576, + "completedOperations": 1, + "durationMillis": 1000, + "errorCount": 2 + } + }, + "expected": { + "accepted": false, + "reason": "INVALID_MEASUREMENT" + } + }, + { + "id": "performance.object.zero-data-duration", + "schemaValid": false, + "value": { + "schemaVersion": 1, + "runUid": "019e3ae0-0000-7000-8000-000000000001", + "toolId": "performance.object", + "capability": "performance.object@1", + "outcome": "SUCCEEDED", + "reasonCode": "COMPLETE", + "durationMillis": 1000, + "provenance": { + "repository": "rustfs/rustfs", + "sourceCommit": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "executableSha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "rustfsVersion": "1.0.0-rc.6", + "osFamily": "LINUX", + "architecture": "aarch64", + "buildFeatures": [] + }, + "coverage": { + "requestedUnits": 1, + "completedUnits": 1, + "unit": "WINDOW" + }, + "data": { + "operation": "GET_OBJECT", + "transferredBytes": 1048576, + "completedOperations": 1, + "durationMillis": 0, + "errorCount": 0 + } + }, + "expected": { + "accepted": false, + "reason": "INVALID_MEASUREMENT" + } + }, + { + "id": "performance.object.unknown-operation", + "schemaValid": false, + "value": { + "schemaVersion": 1, + "runUid": "019e3ae0-0000-7000-8000-000000000001", + "toolId": "performance.object", + "capability": "performance.object@1", + "outcome": "SUCCEEDED", + "reasonCode": "COMPLETE", + "durationMillis": 1000, + "provenance": { + "repository": "rustfs/rustfs", + "sourceCommit": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "executableSha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "rustfsVersion": "1.0.0-rc.6", + "osFamily": "LINUX", + "architecture": "aarch64", + "buildFeatures": [] + }, + "coverage": { + "requestedUnits": 1, + "completedUnits": 1, + "unit": "WINDOW" + }, + "data": { + "operation": "DELETE_OBJECT", + "transferredBytes": 1048576, + "completedOperations": 1, + "durationMillis": 1000, + "errorCount": 0 + } + }, + "expected": { + "accepted": false, + "reason": "UNKNOWN_OPERATION" + } + }, + { + "id": "performance.object.customer-namespace-field", + "schemaValid": false, + "value": { + "schemaVersion": 1, + "runUid": "019e3ae0-0000-7000-8000-000000000001", + "toolId": "performance.object", + "capability": "performance.object@1", + "outcome": "SUCCEEDED", + "reasonCode": "COMPLETE", + "durationMillis": 1000, + "provenance": { + "repository": "rustfs/rustfs", + "sourceCommit": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "executableSha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "rustfsVersion": "1.0.0-rc.6", + "osFamily": "LINUX", + "architecture": "aarch64", + "buildFeatures": [] + }, + "coverage": { + "requestedUnits": 1, + "completedUnits": 1, + "unit": "WINDOW" + }, + "data": { + "operation": "GET_OBJECT", + "transferredBytes": 1048576, + "completedOperations": 1, + "durationMillis": 1000, + "errorCount": 0, + "bucket": "customer-bucket" + } + }, + "expected": { + "accepted": false, + "reason": "UNKNOWN_FIELD" + } + }, + { + "id": "performance.object.credential-field", + "schemaValid": false, + "value": { + "schemaVersion": 1, + "runUid": "019e3ae0-0000-7000-8000-000000000001", + "toolId": "performance.object", + "capability": "performance.object@1", + "outcome": "SUCCEEDED", + "reasonCode": "COMPLETE", + "durationMillis": 1000, + "provenance": { + "repository": "rustfs/rustfs", + "sourceCommit": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "executableSha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "rustfsVersion": "1.0.0-rc.6", + "osFamily": "LINUX", + "architecture": "aarch64", + "buildFeatures": [] + }, + "coverage": { + "requestedUnits": 1, + "completedUnits": 1, + "unit": "WINDOW" + }, + "data": { + "operation": "GET_OBJECT", + "transferredBytes": 1048576, + "completedOperations": 1, + "durationMillis": 1000, + "errorCount": 0, + "accessKey": "SYNTHETIC_FORBIDDEN_CREDENTIAL" + } + }, + "expected": { + "accepted": false, + "reason": "UNKNOWN_FIELD" + } + } + ] +} diff --git a/rustfs/src/config/cli.rs b/rustfs/src/config/cli.rs index 3589c0b0f..1a646a827 100644 --- a/rustfs/src/config/cli.rs +++ b/rustfs/src/config/cli.rs @@ -248,6 +248,8 @@ pub struct ConnectPerformanceOpts { pub enum ConnectPerformanceCommands { /// Measure bounded client-to-deployment transfer performance Client(Box), + /// Measure bounded S3 object throughput in a dedicated temporary namespace + Object(Box), /// Measure generated-file write and warm page-cache read performance Drive(Box), } @@ -357,6 +359,111 @@ pub struct ConnectClientPerformanceOpts { pub acknowledge_l1: bool, } +#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)] +pub enum ConnectObjectPerformanceOperation { + Get, + Put, +} + +#[derive(Args, Clone)] +pub struct ConnectObjectPerformanceOpts { + /// 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, + + /// Optional explicit HTTP(S) proxy without embedded credentials + #[arg(long, value_parser = NonEmptyStringValueParser::new())] + pub proxy: Option, + + /// 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, + + /// 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.object@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, + + /// Object transfer operation + #[arg(long, value_enum)] + pub operation: ConnectObjectPerformanceOperation, + + /// 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)] pub struct ConnectDrivePerformanceOpts { /// Directory containing an enrolled Connect device identity @@ -1096,6 +1203,8 @@ pub enum CommandResult { ConnectDrivePerformance(ConnectDrivePerformanceOpts), /// Consent-bound client-to-deployment performance export ConnectClientPerformance(ConnectClientPerformanceOpts), + /// Consent-bound S3 object performance export + ConnectObjectPerformance(ConnectObjectPerformanceOpts), /// Consent-bound local Connect profile export ConnectProfile(ConnectProfileOpts), /// Consent-bound local Connect log export diff --git a/rustfs/src/config/mod.rs b/rustfs/src/config/mod.rs index 2a4ac6efb..aa262c216 100644 --- a/rustfs/src/config/mod.rs +++ b/rustfs/src/config/mod.rs @@ -57,6 +57,7 @@ pub use cli::{ pub use cli::{ConnectEnvironmentInventoryOpts, ConnectInventoryCommands}; pub use cli::{ConnectLicenseArtifactOpts, ConnectLicenseCommands, ConnectLicenseScopeOpts}; pub use cli::{ConnectLogsMode, ConnectLogsOpts}; +pub use cli::{ConnectObjectPerformanceOperation, ConnectObjectPerformanceOpts}; pub use cli::{ConnectProfileOpts, ConnectProfileTool, ConnectThreadProfileScope}; pub use cli::{ ConnectTelemetryArtifactOpts, ConnectTelemetryCommands, ConnectTelemetryOtlpOpts, ConnectTelemetryRecordOpts, diff --git a/rustfs/src/config/opt.rs b/rustfs/src/config/opt.rs index c539a6c37..d382811d6 100644 --- a/rustfs/src/config/opt.rs +++ b/rustfs/src/config/opt.rs @@ -149,6 +149,7 @@ impl Opt { ConnectCommands::Performance(opts) => match opts.command { ConnectPerformanceCommands::Client(opts) => Ok(CommandResult::ConnectClientPerformance(*opts)), ConnectPerformanceCommands::Drive(opts) => Ok(CommandResult::ConnectDrivePerformance(*opts)), + ConnectPerformanceCommands::Object(opts) => Ok(CommandResult::ConnectObjectPerformance(*opts)), }, ConnectCommands::Profile(opts) => Ok(CommandResult::ConnectProfile(opts)), ConnectCommands::Logs(opts) => Ok(CommandResult::ConnectLogs(opts)), diff --git a/rustfs/src/connect/diagnostics/mod.rs b/rustfs/src/connect/diagnostics/mod.rs index e0737add6..890cfaad9 100644 --- a/rustfs/src/connect/diagnostics/mod.rs +++ b/rustfs/src/connect/diagnostics/mod.rs @@ -15,6 +15,7 @@ mod logs; mod perf_client; mod perf_drive; +mod perf_object; mod profile_cpu; mod profile_memory; mod profile_threads; @@ -46,6 +47,15 @@ pub use perf_drive::{ DriveTargetReasonCode, DriveTargetResult, DriveTargetUnits, LocalDriveConsent, SavedDriveExport, SignedDriveExport, measure_drive, save_signed_drive_export, sign_drive_export, validate_drive_limits, }; +pub use perf_object::{ + LocalObjectConsent, MAX_OBJECT_BANDWIDTH_BYTES_PER_SECOND, MAX_OBJECT_DURATION, MAX_OBJECT_RESULT_BYTES, + MAX_OBJECT_TRAFFIC_BYTES, OBJECT_CAPABILITY, OBJECT_SCHEMA_VERSION, OBJECT_TOOL_ID, ObjectDiagnosticResult, + ObjectMeasurement, ObjectOperation, ObjectOutcome, ObjectPerformanceData, ObjectPerformanceError, ObjectPerformanceRequest, + ObjectProbe, ObjectProbeError, ObjectProbeFuture, ObjectProbeMeasurement, ObjectProvenance, ObjectReasonCode, + ObjectTargetParameters, ObjectTargetReasonCode, ObjectTargetResult, ObjectTargetUnits, S3ObjectProbe, SavedObjectExport, + SignedObjectExport, measure_object, read_protected_object_credential, save_signed_object_export, sign_object_export, + validate_object_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, diff --git a/rustfs/src/connect/diagnostics/perf_object.rs b/rustfs/src/connect/diagnostics/perf_object.rs new file mode 100644 index 000000000..198f52ed7 --- /dev/null +++ b/rustfs/src/connect/diagnostics/perf_object.rs @@ -0,0 +1,1345 @@ +// 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. + +//! Consent-bound object-to-deployment performance measurement. +//! +//! The producer creates a dedicated temporary bucket, writes or reads one +//! generated object, and removes the namespace before returning. It never +//! accepts customer bucket or object names and keeps credentials local. + +use std::fs::{self, File, OpenOptions}; +use std::future::Future; +use std::io::{Cursor, Read, Write as _}; +use std::path::Path; +use std::pin::Pin; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use base64_simd::URL_SAFE_NO_PAD; +use bytes::Bytes; +use futures::StreamExt as _; +use p256::ecdsa::{Signature, SigningKey, signature::Signer as _}; +use p256::pkcs8::DecodePrivateKey as _; +use reqwest::{Client, Method, Response, StatusCode, Url}; +use serde::Serialize; +use sha2::{Digest as _, Sha256}; +use thiserror::Error; +use time::{OffsetDateTime, format_description::well_known::Rfc3339}; +use tokio_util::sync::CancellationToken; +use uuid::{Uuid, Variant, Version}; +use zeroize::Zeroizing; +use zip::{CompressionMethod, ZipWriter, write::SimpleFileOptions}; + +use crate::connect::DeviceIdentity; + +pub const OBJECT_SCHEMA_VERSION: u16 = 1; +pub const OBJECT_TOOL_ID: &str = "performance.object"; +pub const OBJECT_CAPABILITY: &str = "performance.object@1"; +pub const MAX_OBJECT_DURATION: Duration = Duration::from_secs(30); +pub const MAX_OBJECT_TRAFFIC_BYTES: u64 = 1_048_576; +pub const MAX_OBJECT_BANDWIDTH_BYTES_PER_SECOND: u64 = 1_048_576; +pub const MAX_OBJECT_RESULT_BYTES: usize = 262_144; + +const MAX_SAFE_INTEGER: u64 = 9_007_199_254_740_991; +const MAX_BUILD_FEATURES: usize = 64; +const MAX_VALIDITY_SECONDS: i64 = 2_592_000; +const MAX_FUTURE_SKEW_SECONDS: i64 = 300; +const MAX_ENVELOPE_BYTES: usize = 16_384; +const MAX_ARCHIVE_BYTES: usize = 524_288; +const MAX_DECOMPRESSED_BYTES: usize = 278_528; +const SIGNATURE_DOMAIN: &[u8] = b"rustfs-diagnostic-envelope-v1\0"; +const ENVELOPE_PATH: &str = "envelope.json"; +const SIGNATURE_PATH: &str = "envelope.sig"; +const RESULT_PATH: &str = "result.json"; +const MAX_OBJECT_RESPONSE_BYTES: usize = 16_384; +const CLEANUP_RESERVE_MAX: Duration = Duration::from_secs(1); +const OUTPUT_MODE: u32 = 0o600; + +static OBJECT_COLLECTOR_ACTIVE: AtomicBool = AtomicBool::new(false); + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum ObjectOperation { + GetObject, + PutObject, +} + +impl ObjectOperation { + pub const fn as_str(self) -> &'static str { + match self { + Self::GetObject => "GET_OBJECT", + Self::PutObject => "PUT_OBJECT", + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum ObjectOutcome { + Succeeded, + Failed, + Cancelled, +} + +impl ObjectOutcome { + pub const fn as_str(self) -> &'static str { + match self { + Self::Succeeded => "SUCCEEDED", + Self::Failed => "FAILED", + Self::Cancelled => "CANCELLED", + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum ObjectReasonCode { + Complete, + SourceUnavailable, + PermissionDenied, + Cancelled, + CollectionFailed, +} + +impl ObjectReasonCode { + pub const fn as_str(self) -> &'static str { + match self { + Self::Complete => "COMPLETE", + Self::SourceUnavailable => "SOURCE_UNAVAILABLE", + Self::PermissionDenied => "PERMISSION_DENIED", + Self::Cancelled => "CANCELLED", + Self::CollectionFailed => "COLLECTION_FAILED", + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum ObjectTargetReasonCode { + Complete, + EndpointUnavailable, + ProxyFailure, + PermissionDenied, + TimedOut, + Cancelled, + NamespaceConflict, + CleanupFailed, + ProtocolFailure, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ObjectTargetParameters { + pub operation: ObjectOperation, + pub requested_bytes: u64, + pub duration_millis: u64, + pub concurrency: u8, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ObjectTargetUnits { + pub bytes: &'static str, + pub duration: &'static str, + pub latency: &'static str, + pub operation_count: &'static str, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ObjectProvenance { + repository: &'static str, + source_commit: String, + executable_sha256: String, + rustfs_version: String, + os_family: ObjectOsFamily, + architecture: ObjectArchitecture, + build_features: Vec, +} + +impl ObjectProvenance { + pub fn new( + source_commit: impl Into, + executable_sha256: impl Into, + rustfs_version: impl Into, + build_features: Vec, + ) -> Self { + Self { + repository: "rustfs/rustfs", + source_commit: source_commit.into(), + executable_sha256: executable_sha256.into(), + rustfs_version: rustfs_version.into(), + os_family: ObjectOsFamily::current(), + architecture: ObjectArchitecture::current(), + build_features, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +enum ObjectOsFamily { + Linux, + Darwin, + Windows, + Freebsd, + Other, +} + +impl ObjectOsFamily { + fn current() -> Self { + match std::env::consts::OS { + "linux" => Self::Linux, + "macos" => Self::Darwin, + "windows" => Self::Windows, + "freebsd" => Self::Freebsd, + _ => Self::Other, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "lowercase")] +enum ObjectArchitecture { + #[serde(rename = "x86_64")] + X86_64, + Aarch64, + Other, +} + +impl ObjectArchitecture { + fn current() -> Self { + match std::env::consts::ARCH { + "x86_64" => Self::X86_64, + "aarch64" => Self::Aarch64, + _ => Self::Other, + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct LocalObjectConsent { + pub consent_uid: String, + pub policy_revision: u64, + pub expires_at_unix: i64, + pub confirmed: bool, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ObjectPerformanceRequest { + pub organization_name: String, + pub cluster_name: String, + pub device_name: String, + pub run_uid: String, + pub artifact_uid: String, + pub schema_version: u16, + pub capability: String, + pub consent: LocalObjectConsent, + pub produced_at_unix: i64, + pub expires_at_unix: i64, + pub nonce: [u8; 32], + pub duration: Duration, + pub operation: ObjectOperation, + pub traffic_bytes: u64, + pub target_alias: String, + pub provenance: ObjectProvenance, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ObjectPerformanceData { + pub operation: ObjectOperation, + pub transferred_bytes: u64, + pub completed_operations: u64, + pub duration_millis: u64, + pub error_count: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +struct ObjectCoverage { + requested_units: u32, + completed_units: u32, + unit: &'static str, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ObjectDiagnosticResult { + schema_version: u16, + run_uid: String, + tool_id: &'static str, + capability: &'static str, + outcome: ObjectOutcome, + reason_code: ObjectReasonCode, + duration_millis: u64, + provenance: ObjectProvenance, + coverage: ObjectCoverage, + data: Option, +} + +impl ObjectDiagnosticResult { + pub fn outcome(&self) -> ObjectOutcome { + self.outcome + } + + pub fn reason_code(&self) -> ObjectReasonCode { + self.reason_code + } + + pub fn data(&self) -> Option<&ObjectPerformanceData> { + self.data.as_ref() + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ObjectTargetResult { + pub target_alias: String, + pub outcome: ObjectOutcome, + pub reason_code: ObjectTargetReasonCode, + pub parameters: ObjectTargetParameters, + pub units: ObjectTargetUnits, + pub transferred_bytes: u64, + pub completed_operations: u64, + pub latency_micros: Option, + pub duration_millis: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ObjectMeasurement { + pub result: ObjectDiagnosticResult, + pub target: ObjectTargetResult, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ObjectProbeMeasurement { + pub transferred_bytes: u64, + pub duration: Duration, + pub latency: Duration, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ObjectProbeError { + EndpointUnavailable, + ProxyFailure, + PermissionDenied, + TimedOut, + Cancelled, + NamespaceConflict, + CleanupFailed, + ProtocolFailure, +} + +pub type ObjectProbeFuture<'a> = Pin> + Send + 'a>>; + +pub trait ObjectProbe: Send + Sync { + fn probe<'a>(&'a self, request: &'a ObjectPerformanceRequest, cancel: &'a CancellationToken) -> ObjectProbeFuture<'a>; +} + +pub struct S3ObjectProbe { + endpoint: Url, + client: Client, + access_key: Zeroizing, + secret_key: Zeroizing, + session_token: Zeroizing, + proxy_configured: bool, +} + +impl S3ObjectProbe { + pub fn new( + endpoint: &str, + root_ca_pem: Option<&[u8]>, + proxy: Option<&str>, + access_key: Zeroizing, + secret_key: Zeroizing, + session_token: Zeroizing, + timeout: Duration, + ) -> Result { + let endpoint = deployment_endpoint(endpoint)?; + if access_key.is_empty() || secret_key.is_empty() { + return Err(ObjectPerformanceError::InvalidCredential); + } + let mut builder = Client::builder() + .no_proxy() + .redirect(reqwest::redirect::Policy::none()) + .timeout(timeout); + if let Some(root_ca_pem) = root_ca_pem { + let certificate = + reqwest::Certificate::from_pem(root_ca_pem).map_err(|_| ObjectPerformanceError::InvalidRootCertificate)?; + builder = builder.add_root_certificate(certificate); + } + if let Some(proxy) = proxy { + let proxy_url = proxy_url(proxy)?; + builder = builder.proxy(reqwest::Proxy::all(proxy_url).map_err(|_| ObjectPerformanceError::InvalidProxy)?); + } + let client = builder.build().map_err(|_| ObjectPerformanceError::TransportConfiguration)?; + Ok(Self { + endpoint, + client, + access_key, + secret_key, + session_token, + proxy_configured: proxy.is_some(), + }) + } + + async fn execute( + &self, + request: &ObjectPerformanceRequest, + cancel: &CancellationToken, + ) -> Result { + let started = Instant::now(); + let reserve = CLEANUP_RESERVE_MAX.min(request.duration / 4); + let operation_deadline = started + request.duration.saturating_sub(reserve); + let cleanup_deadline = started + request.duration; + let bucket = format!("rustfs-connect-perf-{}", request.artifact_uid.replace('-', "")); + let object = "synthetic-object"; + let bucket_url = self.object_url(&bucket, None)?; + let object_url = self.object_url(&bucket, Some(object))?; + let payload_len = usize::try_from(request.traffic_bytes).map_err(|_| ObjectProbeError::ProtocolFailure)?; + let payload = Bytes::from(vec![0xa5; payload_len]); + + let create = self + .send(Method::PUT, bucket_url.clone(), Bytes::new(), operation_deadline, Some(cancel)) + .await?; + if create.status() == StatusCode::CONFLICT { + return Err(ObjectProbeError::NamespaceConflict); + } + if !create.status().is_success() { + return self.status_error(create.status()); + } + let measurement = match self.require_success(create, operation_deadline, Some(cancel)).await { + Ok(()) => { + self.measure_in_namespace(request, object_url.clone(), payload, operation_deadline, cancel) + .await + } + Err(error) => Err(error), + }; + let cleanup = self.cleanup(object_url, bucket_url, cleanup_deadline).await; + if cleanup.is_err() { + return Err(ObjectProbeError::CleanupFailed); + } + measurement.map(|(transferred_bytes, latency)| ObjectProbeMeasurement { + transferred_bytes, + duration: latency, + latency, + }) + } + + async fn measure_in_namespace( + &self, + request: &ObjectPerformanceRequest, + object_url: Url, + payload: Bytes, + deadline: Instant, + cancel: &CancellationToken, + ) -> Result<(u64, Duration), ObjectProbeError> { + if request.operation == ObjectOperation::GetObject { + let preload = self + .send(Method::PUT, object_url.clone(), payload.clone(), deadline, Some(cancel)) + .await?; + self.require_success(preload, deadline, Some(cancel)).await?; + } + + let operation_started = Instant::now(); + match request.operation { + ObjectOperation::PutObject => { + let response = self.send(Method::PUT, object_url, payload, deadline, Some(cancel)).await?; + self.require_success(response, deadline, Some(cancel)).await?; + } + ObjectOperation::GetObject => { + let response = self + .send(Method::GET, object_url, Bytes::new(), deadline, Some(cancel)) + .await?; + if !response.status().is_success() { + return self.status_error(response.status()); + } + let body = self + .read_response_body(response, request.traffic_bytes, deadline, Some(cancel)) + .await?; + if body.len() != usize::try_from(request.traffic_bytes).map_err(|_| ObjectProbeError::ProtocolFailure)? + || body.iter().any(|byte| *byte != 0xa5) + { + return Err(ObjectProbeError::ProtocolFailure); + } + } + } + Ok((request.traffic_bytes, operation_started.elapsed())) + } + + async fn cleanup(&self, object_url: Url, bucket_url: Url, deadline: Instant) -> Result<(), ObjectProbeError> { + let object = self.send(Method::DELETE, object_url, Bytes::new(), deadline, None).await?; + if !object.status().is_success() && object.status() != StatusCode::NOT_FOUND { + return Err(ObjectProbeError::CleanupFailed); + } + let bucket = self.send(Method::DELETE, bucket_url, Bytes::new(), deadline, None).await?; + if !bucket.status().is_success() { + return Err(ObjectProbeError::CleanupFailed); + } + Ok(()) + } + + async fn send( + &self, + method: Method, + url: Url, + payload: Bytes, + deadline: Instant, + cancel: Option<&CancellationToken>, + ) -> Result { + let payload_hash = hex_lower(&Sha256::digest(&payload)); + let unsigned = http::Request::builder() + .method(method.clone()) + .uri(url.as_str()) + .header("x-amz-content-sha256", payload_hash) + .body(()) + .map_err(|_| ObjectProbeError::ProtocolFailure)?; + let signed_headers = rustfs_signer::try_sign_v4_headers( + unsigned.into_parts().0, + i64::try_from(payload.len()).map_err(|_| ObjectProbeError::ProtocolFailure)?, + &self.access_key, + &self.secret_key, + &self.session_token, + "us-east-1", + ) + .map_err(|_| ObjectProbeError::ProtocolFailure)?; + let send = self.client.request(method, url).headers(signed_headers).body(payload).send(); + let timed = tokio::time::timeout_at(tokio::time::Instant::from_std(deadline), send); + let response = if let Some(cancel) = cancel { + tokio::select! { + () = cancel.cancelled() => return Err(ObjectProbeError::Cancelled), + response = timed => response, + } + } else { + timed.await + }; + response + .map_err(|_| ObjectProbeError::TimedOut)? + .map_err(|error| self.transport_error(&error)) + } + + async fn require_success( + &self, + response: Response, + deadline: Instant, + cancel: Option<&CancellationToken>, + ) -> Result<(), ObjectProbeError> { + if !response.status().is_success() { + return self.status_error(response.status()); + } + self.read_response_body(response, MAX_OBJECT_RESPONSE_BYTES as u64, deadline, cancel) + .await + .map(|_| ()) + } + + async fn read_response_body( + &self, + response: Response, + max_bytes: u64, + deadline: Instant, + cancel: Option<&CancellationToken>, + ) -> Result, ObjectProbeError> { + if response.content_length().is_some_and(|length| length > max_bytes) { + return Err(ObjectProbeError::ProtocolFailure); + } + let max_bytes = usize::try_from(max_bytes).map_err(|_| ObjectProbeError::ProtocolFailure)?; + let mut body = Vec::with_capacity(max_bytes.min(MAX_OBJECT_RESPONSE_BYTES)); + let mut stream = response.bytes_stream(); + loop { + let next = tokio::time::timeout_at(tokio::time::Instant::from_std(deadline), stream.next()); + let chunk = if let Some(cancel) = cancel { + tokio::select! { + () = cancel.cancelled() => return Err(ObjectProbeError::Cancelled), + chunk = next => chunk, + } + } else { + next.await + } + .map_err(|_| ObjectProbeError::TimedOut)?; + let Some(chunk) = chunk else { + break; + }; + let chunk = chunk.map_err(|error| self.transport_error(&error))?; + let length = body.len().checked_add(chunk.len()).ok_or(ObjectProbeError::ProtocolFailure)?; + if length > max_bytes { + return Err(ObjectProbeError::ProtocolFailure); + } + body.extend_from_slice(&chunk); + } + Ok(body) + } + + fn object_url(&self, bucket: &str, object: Option<&str>) -> Result { + let path = object.map_or_else(|| format!("{bucket}/"), |object| format!("{bucket}/{object}")); + self.endpoint.join(&path).map_err(|_| ObjectProbeError::ProtocolFailure) + } + + fn status_error(&self, status: StatusCode) -> Result { + if status == StatusCode::UNAUTHORIZED || status == StatusCode::FORBIDDEN { + Err(ObjectProbeError::PermissionDenied) + } else { + Err(ObjectProbeError::ProtocolFailure) + } + } + + fn transport_error(&self, error: &reqwest::Error) -> ObjectProbeError { + if error.is_timeout() { + ObjectProbeError::TimedOut + } else if self.proxy_configured { + ObjectProbeError::ProxyFailure + } else if error.is_connect() { + ObjectProbeError::EndpointUnavailable + } else { + ObjectProbeError::ProtocolFailure + } + } +} + +impl ObjectProbe for S3ObjectProbe { + fn probe<'a>(&'a self, request: &'a ObjectPerformanceRequest, cancel: &'a CancellationToken) -> ObjectProbeFuture<'a> { + Box::pin(self.execute(request, cancel)) + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SignedObjectExport { + pub artifact_uid: String, + pub outcome: ObjectOutcome, + pub reason_code: ObjectReasonCode, + pub envelope_json: Vec, + pub envelope_signature: Vec, + pub result_json: Vec, + pub archive_bytes: Vec, + pub archive_sha256: String, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SavedObjectExport { + pub artifact_uid: String, + pub archive_size_bytes: u64, + pub archive_sha256: String, +} + +#[derive(Debug, Error)] +pub enum ObjectPerformanceError { + #[error("object_performance_local_consent_required")] + ConsentRequired, + #[error("object_performance_local_consent_expired")] + ConsentExpired, + #[error("object_performance_request_expired")] + Expired, + #[error("object_performance_invalid_request")] + InvalidRequest, + #[error("object_performance_unsupported_version")] + UnsupportedVersion, + #[error("object_performance_unsupported_capability")] + UnsupportedCapability, + #[error("object_performance_limit_exceeded")] + LimitExceeded, + #[error("object_performance_collection_cancelled")] + Cancelled, + #[error("object_performance_busy")] + Busy, + #[error("object_performance_invalid_endpoint")] + InvalidEndpoint, + #[error("object_performance_invalid_proxy")] + InvalidProxy, + #[error("object_performance_invalid_root_certificate")] + InvalidRootCertificate, + #[error("object_performance_invalid_credential")] + InvalidCredential, + #[error("object_performance_transport_configuration")] + TransportConfiguration, + #[error("object_performance_signing_failed")] + Signing, + #[error("object_performance_encoding_failed")] + Encoding, + #[error("object_performance_output_exists")] + AlreadyExists, + #[error("object_performance_io_failed")] + Io(#[source] std::io::Error), + #[error("object_performance_output_durability_failed")] + DurabilityAfterCommit(#[source] std::io::Error), +} + +pub async fn measure_object( + request: &ObjectPerformanceRequest, + probe: &impl ObjectProbe, + cancel: &CancellationToken, +) -> Result { + request.validate(unix_now()?)?; + if OBJECT_COLLECTOR_ACTIVE + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_err() + { + return Err(ObjectPerformanceError::Busy); + } + let _guard = ActiveGuard; + if cancel.is_cancelled() { + return Ok(terminal_measurement( + request, + ObjectOutcome::Cancelled, + ObjectReasonCode::Cancelled, + ObjectTargetReasonCode::Cancelled, + Duration::ZERO, + )); + } + let started = Instant::now(); + // The concrete probe owns the task namespace and must observe cancellation + // before returning so its async cleanup cannot be dropped halfway through. + let result = probe.probe(request, cancel).await; + Ok(match result { + Ok(sample) if sample.transferred_bytes == request.traffic_bytes => success_measurement(request, sample), + Ok(_) => failed_measurement(request, started.elapsed(), ObjectTargetReasonCode::ProtocolFailure), + Err(ObjectProbeError::Cancelled) => terminal_measurement( + request, + ObjectOutcome::Cancelled, + ObjectReasonCode::Cancelled, + ObjectTargetReasonCode::Cancelled, + started.elapsed(), + ), + Err(error) => failed_measurement(request, started.elapsed(), target_reason(error)), + }) +} + +pub fn sign_object_export( + request: &ObjectPerformanceRequest, + measurement: &ObjectMeasurement, + key: &DeviceIdentity, + cancel: &CancellationToken, +) -> Result { + request.validate(unix_now()?)?; + check_cancel(cancel)?; + let result = &measurement.result; + if result.outcome != ObjectOutcome::Succeeded + || result.data.is_none() + || result.run_uid != request.run_uid + || result.schema_version != OBJECT_SCHEMA_VERSION + || result.tool_id != OBJECT_TOOL_ID + || result.capability != OBJECT_CAPABILITY + { + return Err(ObjectPerformanceError::InvalidRequest); + } + let result_json = serde_json::to_vec(result).map_err(|_| ObjectPerformanceError::Encoding)?; + if result_json.is_empty() || result_json.len() > MAX_OBJECT_RESULT_BYTES { + return Err(ObjectPerformanceError::LimitExceeded); + } + let device_key_id = hex_lower(&Sha256::digest(key.public_key_der())); + let result_sha256 = hex_lower(&Sha256::digest(&result_json)); + let envelope = ObjectEnvelope { + format_version: "rustfs.connect.diagnosticEnvelope/1", + protocol_version: "v1", + organization_name: &request.organization_name, + cluster_name: &request.cluster_name, + device_name: &request.device_name, + run_uid: &request.run_uid, + artifact_uid: &request.artifact_uid, + tool_id: OBJECT_TOOL_ID, + schema_version: OBJECT_SCHEMA_VERSION, + classification: "L1", + consent_uid: &request.consent.consent_uid, + policy_revision: request.consent.policy_revision, + produced_at: timestamp(request.produced_at_unix)?, + expires_at: timestamp(request.expires_at_unix)?, + nonce: URL_SAFE_NO_PAD.encode_to_string(request.nonce), + device_key_id: &device_key_id, + payload: ObjectPayload { + path: RESULT_PATH, + media_type: "application/json", + size_bytes: u64::try_from(result_json.len()).map_err(|_| ObjectPerformanceError::LimitExceeded)?, + sha256: &result_sha256, + }, + }; + let envelope_json = serde_json::to_vec(&envelope).map_err(|_| ObjectPerformanceError::Encoding)?; + if envelope_json.is_empty() || envelope_json.len() > MAX_ENVELOPE_BYTES { + return Err(ObjectPerformanceError::LimitExceeded); + } + let envelope_signature = signature_document(key, &device_key_id, &envelope_json)?; + let decompressed = result_json + .len() + .checked_add(envelope_json.len()) + .and_then(|size| size.checked_add(envelope_signature.len())) + .ok_or(ObjectPerformanceError::LimitExceeded)?; + if decompressed > MAX_DECOMPRESSED_BYTES { + return Err(ObjectPerformanceError::LimitExceeded); + } + check_cancel(cancel)?; + if unix_now()? >= request.expires_at_unix { + return Err(ObjectPerformanceError::Expired); + } + let archive_bytes = archive(&envelope_json, &envelope_signature, &result_json)?; + if archive_bytes.len() > MAX_ARCHIVE_BYTES { + return Err(ObjectPerformanceError::LimitExceeded); + } + let archive_sha256 = hex_lower(&Sha256::digest(&archive_bytes)); + Ok(SignedObjectExport { + artifact_uid: request.artifact_uid.clone(), + outcome: result.outcome, + reason_code: result.reason_code, + envelope_json, + envelope_signature, + result_json, + archive_bytes, + archive_sha256, + }) +} + +pub fn save_signed_object_export( + output: &Path, + export: &SignedObjectExport, + cancel: &CancellationToken, +) -> Result { + check_cancel(cancel)?; + if !uuid7(&export.artifact_uid) || export.archive_bytes.is_empty() { + return Err(ObjectPerformanceError::InvalidRequest); + } + if export.archive_bytes.len() > MAX_ARCHIVE_BYTES { + return Err(ObjectPerformanceError::LimitExceeded); + } + if hex_lower(&Sha256::digest(&export.archive_bytes)) != export.archive_sha256 { + return Err(ObjectPerformanceError::InvalidRequest); + } + let parent = output + .parent() + .filter(|path| !path.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + let filename = output + .file_name() + .ok_or(ObjectPerformanceError::InvalidRequest)? + .to_string_lossy(); + let temporary = parent.join(format!(".{filename}.{}.partial", export.artifact_uid)); + let mut options = OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt as _; + options.mode(OUTPUT_MODE); + } + let mut file = options.open(&temporary).map_err(map_create_error)?; + let saved = (|| { + file.write_all(&export.archive_bytes).map_err(ObjectPerformanceError::Io)?; + check_cancel(cancel)?; + file.sync_all().map_err(ObjectPerformanceError::Io)?; + check_cancel(cancel)?; + fs::hard_link(&temporary, output).map_err(map_publish_error)?; + fs::remove_file(&temporary).map_err(ObjectPerformanceError::DurabilityAfterCommit)?; + #[cfg(unix)] + File::open(parent) + .and_then(|directory| directory.sync_all()) + .map_err(ObjectPerformanceError::DurabilityAfterCommit)?; + Ok(SavedObjectExport { + artifact_uid: export.artifact_uid.clone(), + archive_size_bytes: u64::try_from(export.archive_bytes.len()).map_err(|_| ObjectPerformanceError::LimitExceeded)?, + archive_sha256: export.archive_sha256.clone(), + }) + })(); + if saved.is_err() { + let _ = fs::remove_file(&temporary); + } + saved +} + +pub fn validate_object_limits( + duration: Duration, + _operation: ObjectOperation, + traffic_bytes: u64, +) -> Result<(), ObjectPerformanceError> { + if duration.is_zero() + || duration.as_millis() == 0 + || duration > MAX_OBJECT_DURATION + || traffic_bytes == 0 + || traffic_bytes > MAX_OBJECT_TRAFFIC_BYTES + { + return Err(ObjectPerformanceError::LimitExceeded); + } + let bandwidth_budget = u64::try_from( + u128::from(MAX_OBJECT_BANDWIDTH_BYTES_PER_SECOND) + .checked_mul(duration.as_millis()) + .ok_or(ObjectPerformanceError::LimitExceeded)? + / 1_000, + ) + .map_err(|_| ObjectPerformanceError::LimitExceeded)?; + if traffic_bytes > bandwidth_budget.max(1) { + return Err(ObjectPerformanceError::LimitExceeded); + } + Ok(()) +} + +pub fn read_protected_object_credential(path: &Path) -> Result, ObjectPerformanceError> { + let metadata = fs::symlink_metadata(path).map_err(ObjectPerformanceError::Io)?; + if metadata.file_type().is_symlink() || !metadata.is_file() { + return Err(ObjectPerformanceError::InvalidCredential); + } + #[cfg(unix)] + { + use std::os::unix::fs::{MetadataExt as _, OpenOptionsExt as _, PermissionsExt as _}; + if metadata.uid() != process_uid() || metadata.permissions().mode() & 0o077 != 0 { + return Err(ObjectPerformanceError::InvalidCredential); + } + let mut options = OpenOptions::new(); + options.read(true).custom_flags(libc::O_CLOEXEC | libc::O_NOFOLLOW); + let mut file = options.open(path).map_err(ObjectPerformanceError::Io)?; + let opened = file.metadata().map_err(ObjectPerformanceError::Io)?; + if opened.dev() != metadata.dev() || opened.ino() != metadata.ino() { + return Err(ObjectPerformanceError::InvalidCredential); + } + read_credential_value(&mut file) + } + #[cfg(not(unix))] + { + let mut file = File::open(path).map_err(ObjectPerformanceError::Io)?; + read_credential_value(&mut file) + } +} + +fn read_credential_value(reader: &mut impl Read) -> Result, ObjectPerformanceError> { + let mut bytes = Vec::with_capacity(256); + reader + .take(4_097) + .read_to_end(&mut bytes) + .map_err(ObjectPerformanceError::Io)?; + if bytes.is_empty() || bytes.len() > 4_096 || bytes.contains(&0) { + return Err(ObjectPerformanceError::InvalidCredential); + } + while matches!(bytes.last(), Some(b'\n' | b'\r')) { + bytes.pop(); + } + let value = String::from_utf8(bytes).map_err(|_| ObjectPerformanceError::InvalidCredential)?; + if value.is_empty() || value.len() > 4_096 || value.trim() != value { + return Err(ObjectPerformanceError::InvalidCredential); + } + Ok(Zeroizing::new(value)) +} + +#[cfg(unix)] +#[allow(unsafe_code)] +fn process_uid() -> u32 { + // SAFETY: geteuid has no pointer arguments or caller preconditions. + unsafe { libc::geteuid() } +} + +impl ObjectPerformanceRequest { + fn validate(&self, now_unix: i64) -> Result<(), ObjectPerformanceError> { + if self.schema_version != OBJECT_SCHEMA_VERSION { + return Err(ObjectPerformanceError::UnsupportedVersion); + } + if self.capability != OBJECT_CAPABILITY { + return Err(ObjectPerformanceError::UnsupportedCapability); + } + if !self.consent.confirmed || self.consent.policy_revision == 0 { + return Err(ObjectPerformanceError::ConsentRequired); + } + if self.consent.expires_at_unix <= now_unix || self.expires_at_unix > self.consent.expires_at_unix { + return Err(ObjectPerformanceError::ConsentExpired); + } + let validity = self + .expires_at_unix + .checked_sub(self.produced_at_unix) + .ok_or(ObjectPerformanceError::Expired)?; + if self.produced_at_unix > now_unix.saturating_add(MAX_FUTURE_SKEW_SECONDS) + || validity <= 0 + || validity > MAX_VALIDITY_SECONDS + || self.expires_at_unix <= now_unix + { + return Err(ObjectPerformanceError::Expired); + } + validate_object_limits(self.duration, self.operation, self.traffic_bytes)?; + if !uuid7(&self.run_uid) + || !uuid7(&self.artifact_uid) + || !uuid7(&self.consent.consent_uid) + || !resource_names_match(self) + || self.target_alias.is_empty() + || self.target_alias.len() > 128 + || !self + .target_alias + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || b"._:-".contains(&byte)) + || !lower_hex_string(&self.provenance.source_commit, 40) + || !lower_hex_string(&self.provenance.executable_sha256, 64) + || !version(&self.provenance.rustfs_version) + || self.provenance.build_features.len() > MAX_BUILD_FEATURES + || !self.provenance.build_features.iter().all(|value| build_feature(value)) + { + return Err(ObjectPerformanceError::InvalidRequest); + } + Ok(()) + } +} + +struct ActiveGuard; + +impl Drop for ActiveGuard { + fn drop(&mut self) { + OBJECT_COLLECTOR_ACTIVE.store(false, Ordering::Release); + } +} + +fn success_measurement(request: &ObjectPerformanceRequest, sample: ObjectProbeMeasurement) -> ObjectMeasurement { + let duration_millis = elapsed_millis(sample.duration); + let data = ObjectPerformanceData { + operation: request.operation, + transferred_bytes: sample.transferred_bytes, + completed_operations: 1, + duration_millis, + error_count: 0, + }; + ObjectMeasurement { + result: result( + request, + ObjectOutcome::Succeeded, + ObjectReasonCode::Complete, + duration_millis, + 1, + Some(data), + ), + target: ObjectTargetResult { + target_alias: request.target_alias.clone(), + outcome: ObjectOutcome::Succeeded, + reason_code: ObjectTargetReasonCode::Complete, + parameters: target_parameters(request), + units: target_units(), + transferred_bytes: sample.transferred_bytes, + completed_operations: 1, + latency_micros: Some(elapsed_micros(sample.latency)), + duration_millis, + }, + } +} + +fn failed_measurement( + request: &ObjectPerformanceRequest, + elapsed: Duration, + target_reason: ObjectTargetReasonCode, +) -> ObjectMeasurement { + let reason = match target_reason { + ObjectTargetReasonCode::EndpointUnavailable => ObjectReasonCode::SourceUnavailable, + ObjectTargetReasonCode::PermissionDenied => ObjectReasonCode::PermissionDenied, + _ => ObjectReasonCode::CollectionFailed, + }; + terminal_measurement(request, ObjectOutcome::Failed, reason, target_reason, elapsed) +} + +fn terminal_measurement( + request: &ObjectPerformanceRequest, + outcome: ObjectOutcome, + reason_code: ObjectReasonCode, + target_reason: ObjectTargetReasonCode, + elapsed: Duration, +) -> ObjectMeasurement { + let duration_millis = elapsed_millis_allow_zero(elapsed); + ObjectMeasurement { + result: result(request, outcome, reason_code, duration_millis, 0, None), + target: ObjectTargetResult { + target_alias: request.target_alias.clone(), + outcome, + reason_code: target_reason, + parameters: target_parameters(request), + units: target_units(), + transferred_bytes: 0, + completed_operations: 0, + latency_micros: None, + duration_millis, + }, + } +} + +fn target_reason(error: ObjectProbeError) -> ObjectTargetReasonCode { + match error { + ObjectProbeError::EndpointUnavailable => ObjectTargetReasonCode::EndpointUnavailable, + ObjectProbeError::ProxyFailure => ObjectTargetReasonCode::ProxyFailure, + ObjectProbeError::PermissionDenied => ObjectTargetReasonCode::PermissionDenied, + ObjectProbeError::TimedOut => ObjectTargetReasonCode::TimedOut, + ObjectProbeError::Cancelled => ObjectTargetReasonCode::Cancelled, + ObjectProbeError::NamespaceConflict => ObjectTargetReasonCode::NamespaceConflict, + ObjectProbeError::CleanupFailed => ObjectTargetReasonCode::CleanupFailed, + ObjectProbeError::ProtocolFailure => ObjectTargetReasonCode::ProtocolFailure, + } +} + +fn target_parameters(request: &ObjectPerformanceRequest) -> ObjectTargetParameters { + ObjectTargetParameters { + operation: request.operation, + requested_bytes: request.traffic_bytes, + duration_millis: elapsed_millis_allow_zero(request.duration), + concurrency: 1, + } +} + +fn target_units() -> ObjectTargetUnits { + ObjectTargetUnits { + bytes: "BYTE", + duration: "MILLISECOND", + latency: "MICROSECOND", + operation_count: "OPERATION", + } +} + +fn result( + request: &ObjectPerformanceRequest, + outcome: ObjectOutcome, + reason_code: ObjectReasonCode, + duration_millis: u64, + completed_units: u32, + data: Option, +) -> ObjectDiagnosticResult { + ObjectDiagnosticResult { + schema_version: OBJECT_SCHEMA_VERSION, + run_uid: request.run_uid.clone(), + tool_id: OBJECT_TOOL_ID, + capability: OBJECT_CAPABILITY, + outcome, + reason_code, + duration_millis: duration_millis.min(30_000), + provenance: request.provenance.clone(), + coverage: ObjectCoverage { + requested_units: 1, + completed_units, + unit: "WINDOW", + }, + data, + } +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct ObjectEnvelope<'a> { + format_version: &'static str, + protocol_version: &'static str, + organization_name: &'a str, + cluster_name: &'a str, + device_name: &'a str, + run_uid: &'a str, + artifact_uid: &'a str, + tool_id: &'static str, + schema_version: u16, + classification: &'static str, + consent_uid: &'a str, + policy_revision: u64, + produced_at: String, + expires_at: String, + nonce: String, + device_key_id: &'a str, + payload: ObjectPayload<'a>, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct ObjectPayload<'a> { + path: &'static str, + media_type: &'static str, + size_bytes: u64, + sha256: &'a str, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct ObjectSignature<'a> { + algorithm: &'static str, + key_id: &'a str, + value: String, +} + +fn signature_document(key: &DeviceIdentity, key_id: &str, envelope: &[u8]) -> Result, ObjectPerformanceError> { + let pkcs8 = key.to_pkcs8_der().map_err(|_| ObjectPerformanceError::Signing)?; + let signing_key = SigningKey::from_pkcs8_der(pkcs8.as_slice()).map_err(|_| ObjectPerformanceError::Signing)?; + let mut input = Vec::with_capacity(SIGNATURE_DOMAIN.len() + envelope.len()); + input.extend_from_slice(SIGNATURE_DOMAIN); + input.extend_from_slice(envelope); + let signature: Signature = signing_key.sign(&input); + serde_json::to_vec(&ObjectSignature { + algorithm: "ES256", + key_id, + value: URL_SAFE_NO_PAD.encode_to_string(signature.normalize_s().to_bytes()), + }) + .map_err(|_| ObjectPerformanceError::Encoding) +} + +fn archive(envelope: &[u8], signature: &[u8], result: &[u8]) -> Result, ObjectPerformanceError> { + let cursor = Cursor::new(Vec::with_capacity(envelope.len() + signature.len() + result.len() + 512)); + let mut writer = ZipWriter::new(cursor); + let options = SimpleFileOptions::DEFAULT + .compression_method(CompressionMethod::Stored) + .unix_permissions(OUTPUT_MODE); + for (name, bytes) in [(ENVELOPE_PATH, envelope), (SIGNATURE_PATH, signature), (RESULT_PATH, result)] { + writer + .start_file(name, options) + .map_err(|_| ObjectPerformanceError::Encoding)?; + writer.write_all(bytes).map_err(ObjectPerformanceError::Io)?; + } + writer + .finish() + .map(|cursor| cursor.into_inner()) + .map_err(|_| ObjectPerformanceError::Encoding) +} + +fn deployment_endpoint(value: &str) -> Result { + let mut url = Url::parse(value).map_err(|_| ObjectPerformanceError::InvalidEndpoint)?; + let local_http = url.scheme() == "http" + && url + .host_str() + .is_some_and(|host| host == "localhost" || host.parse::().is_ok_and(|ip| ip.is_loopback())); + if (url.scheme() != "https" && !local_http) + || url.cannot_be_a_base() + || !url.username().is_empty() + || url.password().is_some() + || url.path() != "/" + || url.query().is_some() + || url.fragment().is_some() + { + return Err(ObjectPerformanceError::InvalidEndpoint); + } + url.set_query(None); + url.set_fragment(None); + if !url.path().ends_with('/') { + url.set_path(&format!("{}/", url.path())); + } + Ok(url) +} + +fn proxy_url(value: &str) -> Result { + let url = Url::parse(value).map_err(|_| ObjectPerformanceError::InvalidProxy)?; + if !matches!(url.scheme(), "http" | "https") + || url.cannot_be_a_base() + || !url.username().is_empty() + || url.password().is_some() + || url.query().is_some() + || url.fragment().is_some() + { + return Err(ObjectPerformanceError::InvalidProxy); + } + Ok(url) +} + +fn resource_names_match(request: &ObjectPerformanceRequest) -> bool { + let Some(organization_uid) = request.organization_name.strip_prefix("organizations/") else { + return false; + }; + if !uuid7(organization_uid) { + return false; + } + let cluster_prefix = format!("{}/clusters/", request.organization_name); + let Some(cluster_uid) = request.cluster_name.strip_prefix(&cluster_prefix) else { + return false; + }; + if !uuid7(cluster_uid) { + return false; + } + let device_prefix = format!("{}/clusterDevices/", request.cluster_name); + request.device_name.strip_prefix(&device_prefix).is_some_and(uuid7) +} + +fn uuid7(value: &str) -> bool { + Uuid::parse_str(value).is_ok_and(|uuid| { + uuid.get_version() == Some(Version::SortRand) && uuid.get_variant() == Variant::RFC4122 && uuid.to_string() == value + }) +} + +fn lower_hex_string(value: &str, length: usize) -> bool { + value.len() == length + && value + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) +} + +fn version(value: &str) -> bool { + if value.is_empty() || value.len() > 64 { + return false; + } + let (core, prerelease) = value + .split_once('-') + .map_or((value, None), |(core, prerelease)| (core, Some(prerelease))); + let mut parts = core.split('.'); + let valid_core = (0..3).all(|_| { + parts + .next() + .is_some_and(|part| !part.is_empty() && part.bytes().all(|byte| byte.is_ascii_digit())) + }) && parts.next().is_none(); + valid_core + && prerelease.is_none_or(|part| { + !part.is_empty() + && part + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'-')) + }) +} + +fn build_feature(value: &str) -> bool { + value.len() <= 64 + && value.as_bytes().split_first().is_some_and(|(first, rest)| { + first.is_ascii_lowercase() + && rest + .iter() + .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'-' | b'_')) + }) +} + +fn elapsed_millis(duration: Duration) -> u64 { + elapsed_millis_allow_zero(duration).max(1) +} + +fn elapsed_millis_allow_zero(duration: Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX).min(MAX_SAFE_INTEGER) +} + +fn elapsed_micros(duration: Duration) -> u64 { + u64::try_from(duration.as_micros()).unwrap_or(u64::MAX).min(MAX_SAFE_INTEGER) +} + +fn unix_now() -> Result { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|_| ObjectPerformanceError::InvalidRequest) + .and_then(|duration| i64::try_from(duration.as_secs()).map_err(|_| ObjectPerformanceError::InvalidRequest)) +} + +fn timestamp(unix: i64) -> Result { + OffsetDateTime::from_unix_timestamp(unix) + .map_err(|_| ObjectPerformanceError::InvalidRequest)? + .format(&Rfc3339) + .map_err(|_| ObjectPerformanceError::Encoding) +} + +fn check_cancel(cancel: &CancellationToken) -> Result<(), ObjectPerformanceError> { + if cancel.is_cancelled() { + Err(ObjectPerformanceError::Cancelled) + } else { + Ok(()) + } +} + +fn hex_lower(bytes: &[u8]) -> String { + let mut encoded = String::with_capacity(bytes.len() * 2); + for byte in bytes { + use std::fmt::Write as _; + let _ = write!(encoded, "{byte:02x}"); + } + encoded +} + +fn map_create_error(error: std::io::Error) -> ObjectPerformanceError { + if error.kind() == std::io::ErrorKind::AlreadyExists { + ObjectPerformanceError::AlreadyExists + } else { + ObjectPerformanceError::Io(error) + } +} + +fn map_publish_error(error: std::io::Error) -> ObjectPerformanceError { + if error.kind() == std::io::ErrorKind::AlreadyExists { + ObjectPerformanceError::AlreadyExists + } else { + ObjectPerformanceError::Io(error) + } +} diff --git a/rustfs/src/connect/mod.rs b/rustfs/src/connect/mod.rs index 73f4d3466..ef34d4147 100644 --- a/rustfs/src/connect/mod.rs +++ b/rustfs/src/connect/mod.rs @@ -72,6 +72,15 @@ pub use diagnostics::{ 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::{ + LocalObjectConsent, MAX_OBJECT_BANDWIDTH_BYTES_PER_SECOND, MAX_OBJECT_DURATION, MAX_OBJECT_RESULT_BYTES, + MAX_OBJECT_TRAFFIC_BYTES, OBJECT_CAPABILITY, OBJECT_SCHEMA_VERSION, OBJECT_TOOL_ID, ObjectDiagnosticResult, + ObjectMeasurement, ObjectOperation, ObjectOutcome, ObjectPerformanceData, ObjectPerformanceError, ObjectPerformanceRequest, + ObjectProbe, ObjectProbeError, ObjectProbeFuture, ObjectProbeMeasurement, ObjectProvenance, ObjectReasonCode, + ObjectTargetParameters, ObjectTargetReasonCode, ObjectTargetResult, ObjectTargetUnits, S3ObjectProbe, SavedObjectExport, + SignedObjectExport, measure_object, read_protected_object_credential, save_signed_object_export, sign_object_export, + validate_object_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, diff --git a/rustfs/src/startup_entrypoint.rs b/rustfs/src/startup_entrypoint.rs index d35309b29..5d2c8903d 100644 --- a/rustfs/src/startup_entrypoint.rs +++ b/rustfs/src/startup_entrypoint.rs @@ -16,8 +16,8 @@ use crate::{ config::{ CommandResult, Config, ConnectClientPerformanceOperation, ConnectClientPerformanceOpts, ConnectDrivePerformanceOpts, ConnectEnvironmentInventoryOpts, ConnectLicenseCommands, ConnectLicenseScopeOpts, ConnectLogsMode, ConnectLogsOpts, - ConnectProfileOpts, ConnectProfileTool, ConnectTelemetryArtifactOpts, ConnectTelemetryCommands, - ConnectThreadProfileScope, ConnectTopCommands, Opt, + ConnectObjectPerformanceOperation, ConnectObjectPerformanceOpts, 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}, @@ -140,6 +140,7 @@ async fn async_main() -> Result<()> { CommandResult::ConnectEnvironmentInventory(options) => return execute_connect_environment_inventory(options).await, 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::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, @@ -775,6 +776,138 @@ async fn execute_connect_client_performance(options: ConnectClientPerformanceOpt Ok(()) } +async fn execute_connect_object_performance(options: ConnectObjectPerformanceOpts) -> Result<()> { + use crate::connect::{ + IdentityStore, LocalObjectConsent, ObjectOperation, ObjectOutcome, ObjectPerformanceRequest, ObjectProvenance, + S3ObjectProbe, measure_object, read_protected_object_credential, save_signed_object_export, sign_object_export, + validate_object_limits, + }; + use rand::{TryRng as _, rngs::SysRng}; + use zeroize::Zeroizing; + + let duration = Duration::from_millis(options.duration_millis); + let operation = match options.operation { + ConnectObjectPerformanceOperation::Get => ObjectOperation::GetObject, + ConnectObjectPerformanceOperation::Put => ObjectOperation::PutObject, + }; + validate_object_limits(duration, operation, 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 object performance requires an enrolled device identity"))?; + let access_key = read_protected_object_credential(&options.access_key_file).map_err(Error::other)?; + let secret_key = read_protected_object_credential(&options.secret_key_file).map_err(Error::other)?; + let session_token = options + .session_token_file + .as_deref() + .map(read_protected_object_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 object root CA exceeds the 1048576-byte limit")); + } + Some(bytes) + } else { + None + }; + let probe = S3ObjectProbe::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 = ObjectPerformanceRequest { + 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: LocalObjectConsent { + 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, + traffic_bytes: options.traffic_bytes, + target_alias: options.target_alias, + provenance: ObjectProvenance::new( + crate::version::build::COMMIT_HASH, + executable_sha256, + env!("CARGO_PKG_VERSION"), + enabled_build_features(), + ), + }; + let cancel = CancellationToken::new(); + let measurement = measure_object(&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.object 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() != ObjectOutcome::Succeeded { + return Err(Error::other(format!( + "object performance collection ended with {}", + measurement.result.outcome().as_str() + ))); + } + + let export = sign_object_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_object_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, diff --git a/rustfs/tests/agent_protocol_fixtures.rs b/rustfs/tests/agent_protocol_fixtures.rs index b4ce4de62..679258042 100644 --- a/rustfs/tests/agent_protocol_fixtures.rs +++ b/rustfs/tests/agent_protocol_fixtures.rs @@ -34,21 +34,22 @@ fn sha256_hex(bytes: &[u8]) -> String { Sha256::digest(bytes).iter().map(|byte| format!("{byte:02x}")).collect() } -/// The registry is closed at eight sets; a ninth is a protocol change, not a +/// The registry is closed; adding another set is a protocol change, not a /// fixture change. Mirrors `EXPECTED_SETS` in Connect's checker. -const EXPECTED_SETS: [&str; 8] = [ +const EXPECTED_SETS: [&str; 9] = [ "auth", "version", "registration", "heartbeat", "inventory", + "object-performance", "offline-enrollment", "bundle", "redaction", ]; #[test] -fn agent_protocol_fixtures_registry_is_the_frozen_eight_sets() { +fn agent_protocol_fixtures_registry_is_the_frozen_sets() { let registry: serde_json::Value = serde_json::from_slice(&fs::read(fixture_root().join("fixture-sets.json")).expect("read fixture-sets.json")) .expect("fixture-sets.json parses"); diff --git a/rustfs/tests/connect_perf_object.rs b/rustfs/tests/connect_perf_object.rs new file mode 100644 index 000000000..8f242e9f7 --- /dev/null +++ b/rustfs/tests/connect_perf_object.rs @@ -0,0 +1,628 @@ +// 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::{ + DeviceIdentity, LocalObjectConsent, OBJECT_CAPABILITY, ObjectOperation, ObjectOutcome, ObjectPerformanceError, + ObjectPerformanceRequest, ObjectProbe, ObjectProbeError, ObjectProbeFuture, ObjectProbeMeasurement, ObjectProvenance, + ObjectReasonCode, ObjectTargetReasonCode, S3ObjectProbe, measure_object, save_signed_object_export, sign_object_export, + validate_object_limits, +}; +use rustfs::embedded::{RustFSServerBuilder, find_available_port}; +use sha2::{Digest as _, Sha256}; +use tokio_util::sync::CancellationToken; +use zeroize::Zeroizing; + +mod common; + +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: ObjectOperation) -> ObjectPerformanceRequest { + let now = now(); + let organization = "organizations/019e3ae0-0000-7000-8000-000000000010"; + let cluster = format!("{organization}/clusters/019e3ae0-0000-7000-8000-000000000011"); + ObjectPerformanceRequest { + 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: OBJECT_CAPABILITY.to_owned(), + consent: LocalObjectConsent { + 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: ObjectProvenance::new("c".repeat(40), "d".repeat(64), "1.0.0-rc.6", vec!["default".to_owned()]), + } +} + +struct SuccessfulProbe; + +impl ObjectProbe for SuccessfulProbe { + fn probe<'a>(&'a self, request: &'a ObjectPerformanceRequest, _cancel: &'a CancellationToken) -> ObjectProbeFuture<'a> { + Box::pin(async move { + Ok(ObjectProbeMeasurement { + transferred_bytes: request.traffic_bytes, + duration: Duration::from_millis(100), + latency: Duration::from_millis(7), + }) + }) + } +} + +struct ErrorProbe(ObjectProbeError); + +impl ObjectProbe for ErrorProbe { + fn probe<'a>(&'a self, _request: &'a ObjectPerformanceRequest, _cancel: &'a CancellationToken) -> ObjectProbeFuture<'a> { + Box::pin(async move { Err(self.0) }) + } +} + +struct PendingProbe; + +impl ObjectProbe for PendingProbe { + fn probe<'a>(&'a self, request: &'a ObjectPerformanceRequest, cancel: &'a CancellationToken) -> ObjectProbeFuture<'a> { + Box::pin(async move { + tokio::select! { + () = cancel.cancelled() => Err(ObjectProbeError::Cancelled), + () = tokio::time::sleep(request.duration) => Err(ObjectProbeError::TimedOut), + } + }) + } +} + +struct CountingProbe(AtomicUsize); + +impl ObjectProbe for CountingProbe { + fn probe<'a>(&'a self, request: &'a ObjectPerformanceRequest, _cancel: &'a CancellationToken) -> ObjectProbeFuture<'a> { + self.0.fetch_add(1, Ordering::Relaxed); + Box::pin(async move { + Ok(ObjectProbeMeasurement { + 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 [ObjectOperation::GetObject, ObjectOperation::PutObject] { + let request = request(operation); + let measurement = measure_object(&request, &SuccessfulProbe, &CancellationToken::new()) + .await + .expect("object measurement"); + assert_eq!( + measurement.result.outcome(), + ObjectOutcome::Succeeded, + "target result: {:?}", + measurement.target + ); + assert_eq!(measurement.result.reason_code(), ObjectReasonCode::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.object"); + assert_eq!(value["capability"], "performance.object@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(ObjectOperation::PutObject); + unsupported_version.schema_version = 2; + assert!(matches!( + measure_object(&unsupported_version, &probe, &CancellationToken::new()).await, + Err(ObjectPerformanceError::UnsupportedVersion) + )); + + let mut unsupported_capability = request(ObjectOperation::PutObject); + unsupported_capability.capability = "performance.object@2".to_owned(); + assert!(matches!( + measure_object(&unsupported_capability, &probe, &CancellationToken::new()).await, + Err(ObjectPerformanceError::UnsupportedCapability) + )); + + let mut no_consent = request(ObjectOperation::PutObject); + no_consent.consent.confirmed = false; + assert!(matches!( + measure_object(&no_consent, &probe, &CancellationToken::new()).await, + Err(ObjectPerformanceError::ConsentRequired) + )); + + let mut over_traffic = request(ObjectOperation::PutObject); + over_traffic.traffic_bytes = 1_048_577; + assert!(matches!( + measure_object(&over_traffic, &probe, &CancellationToken::new()).await, + Err(ObjectPerformanceError::LimitExceeded) + )); + + let mut invalid_version = request(ObjectOperation::PutObject); + invalid_version.provenance = + ObjectProvenance::new("a".repeat(40), "b".repeat(64), "release_candidate", vec!["default".to_owned()]); + assert!(matches!( + measure_object(&invalid_version, &probe, &CancellationToken::new()).await, + Err(ObjectPerformanceError::InvalidRequest) + )); + + let mut invalid_feature = request(ObjectOperation::PutObject); + invalid_feature.provenance = ObjectProvenance::new("a".repeat(40), "b".repeat(64), "1.0.0-rc.6", vec!["Default".to_owned()]); + assert!(matches!( + measure_object(&invalid_feature, &probe, &CancellationToken::new()).await, + Err(ObjectPerformanceError::InvalidRequest) + )); + assert_eq!(probe.0.load(Ordering::Relaxed), 0); + + assert!(validate_object_limits(Duration::from_secs(1), ObjectOperation::PutObject, 1_048_576).is_ok()); + assert!(validate_object_limits(Duration::from_secs(1), ObjectOperation::GetObject, 1_048_576).is_ok()); + assert!(matches!( + validate_object_limits(Duration::from_secs(1), ObjectOperation::PutObject, 1_048_577), + Err(ObjectPerformanceError::LimitExceeded) + )); + assert!(matches!( + validate_object_limits(Duration::from_secs(1), ObjectOperation::GetObject, 1_048_577), + Err(ObjectPerformanceError::LimitExceeded) + )); + assert!(matches!( + validate_object_limits(Duration::from_nanos(1), ObjectOperation::PutObject, 1), + Err(ObjectPerformanceError::LimitExceeded) + )); + assert!(matches!( + validate_object_limits(Duration::from_secs(30) + Duration::from_nanos(1), ObjectOperation::PutObject, 1,), + Err(ObjectPerformanceError::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 [ + ( + ObjectProbeError::EndpointUnavailable, + ObjectTargetReasonCode::EndpointUnavailable, + ObjectReasonCode::SourceUnavailable, + ), + ( + ObjectProbeError::ProxyFailure, + ObjectTargetReasonCode::ProxyFailure, + ObjectReasonCode::CollectionFailed, + ), + ( + ObjectProbeError::PermissionDenied, + ObjectTargetReasonCode::PermissionDenied, + ObjectReasonCode::PermissionDenied, + ), + ( + ObjectProbeError::TimedOut, + ObjectTargetReasonCode::TimedOut, + ObjectReasonCode::CollectionFailed, + ), + ( + ObjectProbeError::NamespaceConflict, + ObjectTargetReasonCode::NamespaceConflict, + ObjectReasonCode::CollectionFailed, + ), + ( + ObjectProbeError::CleanupFailed, + ObjectTargetReasonCode::CleanupFailed, + ObjectReasonCode::CollectionFailed, + ), + ] { + let measurement = measure_object(&request(ObjectOperation::PutObject), &ErrorProbe(error), &CancellationToken::new()) + .await + .expect("typed failure"); + assert_eq!(measurement.result.outcome(), ObjectOutcome::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_object(&request(ObjectOperation::PutObject), &SuccessfulProbe, &cancel) + .await + .expect("typed cancellation"); + assert_eq!(measurement.result.outcome(), ObjectOutcome::Cancelled); + assert_eq!(measurement.target.reason_code, ObjectTargetReasonCode::Cancelled); +} + +#[tokio::test] +async fn deadline_and_in_flight_cancellation_stop_a_stalled_probe() { + let _guard = TEST_LOCK.lock().await; + let mut timed = request(ObjectOperation::GetObject); + timed.duration = Duration::from_millis(20); + timed.traffic_bytes = 1_024; + let measurement = measure_object(&timed, &PendingProbe, &CancellationToken::new()) + .await + .expect("typed timeout"); + assert_eq!(measurement.result.outcome(), ObjectOutcome::Failed); + assert_eq!(measurement.target.reason_code, ObjectTargetReasonCode::TimedOut); + + let cancel = CancellationToken::new(); + let cancellation = cancel.clone(); + tokio::spawn(async move { + tokio::task::yield_now().await; + cancellation.cancel(); + }); + let measurement = measure_object(&request(ObjectOperation::PutObject), &PendingProbe, &cancel) + .await + .expect("typed in-flight cancellation"); + assert_eq!(measurement.result.outcome(), ObjectOutcome::Cancelled); + assert_eq!(measurement.target.reason_code, ObjectTargetReasonCode::Cancelled); +} + +#[tokio::test] +async fn only_one_object_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(ObjectOperation::GetObject); + let second_request = request(ObjectOperation::PutObject); + let first = measure_object(&first_request, &PendingProbe, &first_cancel); + let second = async { + tokio::task::yield_now().await; + measure_object(&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(), ObjectOutcome::Cancelled); + assert!(matches!(second, Err(ObjectPerformanceError::Busy))); +} + +#[tokio::test] +async fn configured_proxy_and_direct_endpoint_failures_are_distinct() { + let _guard = TEST_LOCK.lock().await; + let proxy_probe = S3ObjectProbe::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_object(&request(ObjectOperation::GetObject), &proxy_probe, &CancellationToken::new()) + .await + .expect("typed proxy failure"); + assert_eq!(measurement.target.reason_code, ObjectTargetReasonCode::ProxyFailure); + + let direct_probe = S3ObjectProbe::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_object(&request(ObjectOperation::GetObject), &direct_probe, &CancellationToken::new()) + .await + .expect("typed endpoint failure"); + assert_eq!(measurement.target.reason_code, ObjectTargetReasonCode::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 { + for response in [ + b"HTTP/1.1 200 OK\r\nContent-Length: 65537\r\nConnection: close\r\n\r\n".as_slice(), + b"HTTP/1.1 204 No Content\r\nConnection: close\r\n\r\n".as_slice(), + b"HTTP/1.1 204 No Content\r\nConnection: close\r\n\r\n".as_slice(), + ] { + let (mut socket, _) = listener.accept().await.expect("object connection"); + let mut request = vec![0_u8; 16 * 1024]; + let _ = socket.read(&mut request).await.expect("request headers"); + socket.write_all(response).await.expect("response headers"); + } + }); + let probe = S3ObjectProbe::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("object probe"); + let measurement = measure_object(&request(ObjectOperation::GetObject), &probe, &CancellationToken::new()) + .await + .expect("typed protocol failure"); + assert_eq!(measurement.target.reason_code, ObjectTargetReasonCode::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(ObjectOperation::PutObject); + let measurement = measure_object(&request, &SuccessfulProbe, &CancellationToken::new()) + .await + .expect("object measurement"); + let export = sign_object_export(&request, &measurement, &DeviceIdentity::generate(), &CancellationToken::new()) + .expect("signed export"); + assert!( + !export + .archive_bytes + .windows(b"access".len()) + .any(|window| window == b"access") + ); + assert!( + !export + .archive_bytes + .windows(b"secret".len()) + .any(|window| window == b"secret") + ); + + let failed = measure_object(&request, &ErrorProbe(ObjectProbeError::EndpointUnavailable), &CancellationToken::new()) + .await + .expect("typed failure"); + assert!(matches!( + sign_object_export(&request, &failed, &DeviceIdentity::generate(), &CancellationToken::new()), + Err(ObjectPerformanceError::InvalidRequest) + )); + let directory = tempfile::tempdir().expect("output directory"); + let output = directory.path().join("object.zip"); + let saved = save_signed_object_export(&output, &export, &CancellationToken::new()).expect("save export"); + assert_eq!(saved.archive_sha256, export.archive_sha256); + assert!(matches!( + save_signed_object_export(&output, &export, &CancellationToken::new()), + Err(ObjectPerformanceError::AlreadyExists) + )); + + let mut tampered = export.clone(); + tampered.archive_bytes[0] ^= 0xff; + assert!(matches!( + save_signed_object_export(&directory.path().join("tampered.zip"), &tampered, &CancellationToken::new()), + Err(ObjectPerformanceError::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_object_export(&directory.path().join("oversized.zip"), &oversized, &CancellationToken::new()), + Err(ObjectPerformanceError::LimitExceeded) + )); + + let cancelled = CancellationToken::new(); + cancelled.cancel(); + assert!(matches!( + save_signed_object_export(&directory.path().join("cancelled.zip"), &export, &cancelled), + Err(ObjectPerformanceError::Cancelled) + )); +} + +#[test] +fn real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put() { + common::run_embedded_test(real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put_body); +} + +async fn real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put_body() { + 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("object-perf-access") + .secret_key("object-perf-secret") + .build() + .await + .expect("start embedded server"); + let probe = S3ObjectProbe::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("object probe"); + + for operation in [ObjectOperation::GetObject, ObjectOperation::PutObject] { + let mut request = request(operation); + request.duration = Duration::from_secs(2); + if operation == ObjectOperation::PutObject { + request.artifact_uid = "019e3ae0-0000-7000-8000-000000000016".to_owned(); + } + let measurement = measure_object(&request, &probe, &CancellationToken::new()) + .await + .expect("real object measurement"); + assert_eq!( + measurement.result.outcome(), + ObjectOutcome::Succeeded, + "target result: {:?}", + measurement.target + ); + assert_eq!(measurement.result.data().expect("data").transferred_bytes, 65_536); + assert_eq!(measurement.target.completed_operations, 1); + } + let temp = tempfile::tempdir().expect("CLI tempdir"); + let state = temp.path().join("state"); + let output = temp.path().join("object.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", "object", "--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.object 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>, name: &str) -> Vec { + 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) +}