mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-20 17:43:34 +00:00
feat(connect): add bounded object performance diagnostics (#7734)
feat(connect): add object performance producer
This commit is contained in:
Chris
@@ -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",
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
a499742c06046e9cf2f781fc17d8b4cf95e018ab45fa41ee5944ba15b0e2902c auth
|
||||
9b54e39584a442270ae486e8378df1d6f168c45d0c59baba9459b46bb212d2df version
|
||||
159d116e965b51771f1687fc6ec9bdfbf2a4a0e6d6aa26d3f1046b3dac2697dd registration
|
||||
f707b09b257c7d652eff2a3f680ecb0c8c45ebb4c2defa8dc07f9dd7ee46ca04 heartbeat
|
||||
cc6c3f3bf5e6a938f2e00b8ce42d8fa7bfb974f4eb2927101d4ede4ebdc10dd5 inventory
|
||||
0207458ec9b97b368e2f347da112511a3aabd27d752890cacda4fa2408e7f19e object-performance
|
||||
a2557ab1f7f70fb86a8affc8451596c272478fd2d5500c367b8f9e8385d72578 offline-enrollment
|
||||
04303e0b3c0d1a9de6957555c9ddd6d96b98be2b009de79ce223dd56b81e1909 bundle
|
||||
42cc936dc8fe87335ebff5abe01e3ebf85bdada7f37b43bfd8add46435c9d6f9 redaction
|
||||
@@ -0,0 +1,2 @@
|
||||
24cd96339f27034a0e2bb656ee0c7da9bec9326ded4583a3c9afef06595626b2 accept-vectors.json
|
||||
a9f4376ea5bbf636851e683cdfaca0cb7ebbc9dfce8ffa4dc8b4a12f82307a53 reject-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
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -248,6 +248,8 @@ pub struct ConnectPerformanceOpts {
|
||||
pub enum ConnectPerformanceCommands {
|
||||
/// Measure bounded client-to-deployment transfer performance
|
||||
Client(Box<ConnectClientPerformanceOpts>),
|
||||
/// Measure bounded S3 object throughput in a dedicated temporary namespace
|
||||
Object(Box<ConnectObjectPerformanceOpts>),
|
||||
/// Measure generated-file write and warm page-cache read performance
|
||||
Drive(Box<ConnectDrivePerformanceOpts>),
|
||||
}
|
||||
@@ -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<PathBuf>,
|
||||
|
||||
/// Optional explicit HTTP(S) proxy without embedded credentials
|
||||
#[arg(long, value_parser = NonEmptyStringValueParser::new())]
|
||||
pub proxy: Option<String>,
|
||||
|
||||
/// Owner-readable file containing the S3 access key
|
||||
#[arg(long = "access-key-file")]
|
||||
pub access_key_file: PathBuf,
|
||||
|
||||
/// Owner-readable file containing the S3 secret key
|
||||
#[arg(long = "secret-key-file")]
|
||||
pub secret_key_file: PathBuf,
|
||||
|
||||
/// Optional owner-readable file containing an S3 session token
|
||||
#[arg(long = "session-token-file")]
|
||||
pub session_token_file: Option<PathBuf>,
|
||||
|
||||
/// New local archive path; an existing file is never replaced
|
||||
#[arg(long)]
|
||||
pub output: PathBuf,
|
||||
|
||||
/// Negotiated producer schema version
|
||||
#[arg(long = "schema-version", default_value_t = 1)]
|
||||
pub schema_version: u16,
|
||||
|
||||
/// Negotiated producer capability
|
||||
#[arg(long, default_value = "performance.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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)),
|
||||
|
||||
@@ -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,
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -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<Cursor<&[u8]>>, name: &str) -> Vec<u8> {
|
||||
let mut bytes = Vec::new();
|
||||
archive
|
||||
.by_name(name)
|
||||
.expect("archive member")
|
||||
.read_to_end(&mut bytes)
|
||||
.expect("read archive member");
|
||||
bytes
|
||||
}
|
||||
|
||||
fn sha256_file(path: &Path) -> String {
|
||||
let mut file = File::open(path).expect("open exact binary");
|
||||
let mut digest = Sha256::new();
|
||||
let mut buffer = [0_u8; 64 * 1024];
|
||||
loop {
|
||||
let read = file.read(&mut buffer).expect("hash exact binary");
|
||||
if read == 0 {
|
||||
break;
|
||||
}
|
||||
digest.update(&buffer[..read]);
|
||||
}
|
||||
hex_lower(&digest.finalize())
|
||||
}
|
||||
|
||||
fn hex_lower(bytes: &[u8]) -> String {
|
||||
hex_simd::encode_to_string(bytes, hex_simd::AsciiCase::Lower)
|
||||
}
|
||||
Reference in New Issue
Block a user