mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-12 15:48:35 +00:00
feat(connect): add bounded health service job
This commit is contained in:
Chris
@@ -110,6 +110,12 @@ use uuid::Uuid;
|
||||
// Bound outstanding filesystem jobs and metadata buffers per disk request.
|
||||
const PART_METADATA_READ_CONCURRENCY: usize = 8;
|
||||
|
||||
/// The fresh health path uses a synchronous capacity syscall. Keep the permit
|
||||
/// inside the blocking closure so cancelling its async caller cannot admit a
|
||||
/// second probe while the first syscall is still running.
|
||||
static FRESH_CAPACITY_PROBE_PERMIT: std::sync::LazyLock<Arc<Semaphore>> =
|
||||
std::sync::LazyLock::new(|| Arc::new(Semaphore::new(1)));
|
||||
|
||||
const DELETED_OBJECTS_CLEANUP_INTERVAL: Duration = Duration::from_secs(60 * 5);
|
||||
const STALE_TMP_OBJECT_EXPIRY: Duration = Duration::from_secs(24 * 60 * 60);
|
||||
|
||||
@@ -11123,7 +11129,34 @@ impl DiskAPI for LocalDisk {
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
async fn disk_info(&self, _: &DiskInfoOptions) -> Result<DiskInfo> {
|
||||
async fn disk_info(&self, opts: &DiskInfoOptions) -> Result<DiskInfo> {
|
||||
if opts.fresh_capacity {
|
||||
let permit = FRESH_CAPACITY_PROBE_PERMIT
|
||||
.clone()
|
||||
.acquire_owned()
|
||||
.await
|
||||
.map_err(|_| DiskError::other("fresh capacity probe unavailable"))?;
|
||||
let root = self.root.clone();
|
||||
let info = tokio::task::spawn_blocking(move || {
|
||||
let _permit = permit;
|
||||
let drive_path = root.to_string_lossy().to_string();
|
||||
check_path_length(&drive_path)?;
|
||||
get_info(&drive_path).map_err(DiskError::from)
|
||||
})
|
||||
.await
|
||||
.map_err(|_| DiskError::other("fresh capacity probe failed"))??;
|
||||
let healing = fs::try_exists(self.root.join(RUSTFS_META_BUCKET).join(super::HEALING_MARKER_PATH))
|
||||
.await
|
||||
.unwrap_or(false);
|
||||
return Ok(DiskInfo {
|
||||
total: info.total,
|
||||
free: info.free,
|
||||
used: info.used,
|
||||
healing,
|
||||
fresh_capacity: true,
|
||||
..Default::default()
|
||||
});
|
||||
}
|
||||
let mut info = Cache::get(self.disk_info_cache.clone()).await?;
|
||||
info.nr_requests = self.nrrequests;
|
||||
info.rotational = self.rotational;
|
||||
@@ -21108,6 +21141,7 @@ mod test {
|
||||
disk_id: "test-disk".to_string(),
|
||||
metrics: true,
|
||||
noop: false,
|
||||
fresh_capacity: false,
|
||||
};
|
||||
|
||||
let disk_info = disk.disk_info(&disk_info_opts).await.expect("operation should succeed");
|
||||
@@ -21125,6 +21159,26 @@ mod test {
|
||||
assert!(!disk_info.mount_path.is_empty());
|
||||
assert!(!disk_info.endpoint.is_empty());
|
||||
|
||||
let fresh = disk
|
||||
.disk_info(&DiskInfoOptions {
|
||||
fresh_capacity: true,
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.expect("fresh capacity should be available");
|
||||
assert!(fresh.fresh_capacity);
|
||||
assert!(fresh.total > 0);
|
||||
assert!(fresh.used <= fresh.total);
|
||||
assert!(fresh.endpoint.is_empty());
|
||||
assert!(fresh.mount_path.is_empty());
|
||||
assert!(fresh.physical_device_ids.is_empty());
|
||||
assert!(fresh.id.is_none());
|
||||
assert!(fresh.fs_type.is_empty());
|
||||
assert_eq!(fresh.major, 0);
|
||||
assert_eq!(fresh.minor, 0);
|
||||
assert_eq!(fresh.used_inodes, 0);
|
||||
assert_eq!(fresh.free_inodes, 0);
|
||||
|
||||
// Clean up the test directory
|
||||
let _ = fs::remove_dir_all(&test_dir).await;
|
||||
}
|
||||
|
||||
@@ -1365,6 +1365,10 @@ pub struct DiskInfoOptions {
|
||||
pub disk_id: String,
|
||||
pub metrics: bool,
|
||||
pub noop: bool,
|
||||
/// Bypass the ordinary one-second admin cache and return a positive
|
||||
/// acknowledgement only after a fresh capacity syscall completes.
|
||||
#[serde(default)]
|
||||
pub fresh_capacity: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
|
||||
@@ -1389,6 +1393,10 @@ pub struct DiskInfo {
|
||||
pub rotational: bool,
|
||||
pub metrics: DiskMetrics,
|
||||
pub error: String,
|
||||
/// Older peers omit this field. Callers must treat an absent/false value as
|
||||
/// cached or otherwise unproven capacity evidence.
|
||||
#[serde(default)]
|
||||
pub fresh_capacity: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default)]
|
||||
@@ -1912,11 +1920,40 @@ mod tests {
|
||||
disk_id: "test-disk-id".to_string(),
|
||||
metrics: true,
|
||||
noop: false,
|
||||
fresh_capacity: true,
|
||||
};
|
||||
|
||||
assert_eq!(opts.disk_id, "test-disk-id");
|
||||
assert!(opts.metrics);
|
||||
assert!(!opts.noop);
|
||||
assert!(opts.fresh_capacity);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn disk_info_fresh_capacity_ack_is_additive_for_old_peers() {
|
||||
let old_options: DiskInfoOptions =
|
||||
serde_json::from_str(r#"{"disk_id":"disk","metrics":false,"noop":false}"#).expect("old options");
|
||||
assert!(!old_options.fresh_capacity);
|
||||
|
||||
let mut old_response_json = serde_json::to_value(DiskInfo {
|
||||
total: 10,
|
||||
free: 4,
|
||||
used: 6,
|
||||
..Default::default()
|
||||
})
|
||||
.expect("response fixture");
|
||||
old_response_json
|
||||
.as_object_mut()
|
||||
.expect("response object")
|
||||
.remove("fresh_capacity");
|
||||
let old_response: DiskInfo = serde_json::from_value(old_response_json).expect("old response");
|
||||
assert!(!old_response.fresh_capacity);
|
||||
|
||||
let requested = DiskInfoOptions {
|
||||
fresh_capacity: true,
|
||||
..Default::default()
|
||||
};
|
||||
assert_eq!(serde_json::to_value(requested).expect("new options")["fresh_capacity"], true);
|
||||
}
|
||||
|
||||
/// Test ReadMultipleReq structure
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
29c93b4c8c27164896c5197e1df4ba03175425ccf5b12dc8e3f3d8d6bf19dc51 accept-vectors.json
|
||||
9423c9fd02b47940cfe54915e0f90706469de2111176053d8ac0c86f30e443ff health-service-accept-vectors.json
|
||||
166fbfb022ab7e6780547b21c53ca36f357e74b56355c84c8174393449942dfa health-service-reject-vectors.json
|
||||
fef527fdd89aa87fa9f81394eab6484fc2ac05a31559e1589149716193b336fb receipt-vectors.json
|
||||
219749ea7bdf73461108320e13ba14b604a1967e32139c5317a017ad95dab000 reject-vectors.json
|
||||
|
||||
@@ -0,0 +1,123 @@
|
||||
{
|
||||
"protocolVersion": "v1",
|
||||
"fixtureSet": "diagnostic-scheduler",
|
||||
"fixture": "health-service-accept-vectors",
|
||||
"vectors": [
|
||||
{
|
||||
"id": "health-service.job.explicit-freshness",
|
||||
"schema": "job-envelope.schema.json",
|
||||
"value": {
|
||||
"jobId": "018cc251-f400-7abc-8def-0123456789ab",
|
||||
"protocolVersion": "v1",
|
||||
"jobType": "health.check",
|
||||
"schemaVersion": 1,
|
||||
"organizationName": "organizations/018cc251-f400-7abc-8def-0123456789ab",
|
||||
"clusterName": "organizations/018cc251-f400-7abc-8def-0123456789ab/clusters/018cc251-f400-7abc-8def-0123456789ac",
|
||||
"deviceName": "organizations/018cc251-f400-7abc-8def-0123456789ab/clusters/018cc251-f400-7abc-8def-0123456789ac/clusterDevices/018cc251-f400-7abc-8def-0123456789ad",
|
||||
"createTime": "2030-01-01T00:00:00Z",
|
||||
"expireTime": "2030-01-01T00:00:30Z",
|
||||
"nonce": "BwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwc",
|
||||
"requiredCapabilities": ["health.check.service@1"],
|
||||
"authorization": {
|
||||
"actorType": "BROWSER_USER",
|
||||
"actorName": "users/018cc251-f400-7abc-8def-0123456789ab",
|
||||
"requestId": "123e4567-e89b-42d3-a456-426614174001"
|
||||
},
|
||||
"limits": {
|
||||
"timeoutSeconds": 30,
|
||||
"maxOutputBytes": 262144,
|
||||
"maxMemoryBytes": 67108864,
|
||||
"maxCpuMillis": 5000
|
||||
},
|
||||
"parameters": {
|
||||
"artifactUid": "018cc251-f400-7abc-8def-0123456789ae",
|
||||
"consentUid": "018cc251-f400-7abc-8def-0123456789af",
|
||||
"consentPolicyRevision": 1,
|
||||
"consentExpiresAt": "2030-01-01T00:01:00Z",
|
||||
"maxEvidenceAgeSeconds": 300
|
||||
},
|
||||
"signature": {
|
||||
"algorithm": "Ed25519",
|
||||
"keyId": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
|
||||
"value": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"id": "health-service.result.current-partial-catalog",
|
||||
"schema": "health-service-result.schema.json",
|
||||
"value": {
|
||||
"schemaVersion": 1,
|
||||
"runUid": "018cc251-f400-7abc-8def-0123456789ab",
|
||||
"toolId": "health.check",
|
||||
"capability": "health.check.service@1",
|
||||
"outcome": "PARTIAL",
|
||||
"reasonCode": "CATALOG_PARTIAL",
|
||||
"durationMillis": 25,
|
||||
"provenance": {
|
||||
"repository": "rustfs/rustfs",
|
||||
"sourceCommit": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
|
||||
"executableSha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
|
||||
"rustfsVersion": "1.0.0",
|
||||
"osFamily": "LINUX",
|
||||
"architecture": "x86_64",
|
||||
"buildFeatures": []
|
||||
},
|
||||
"coverage": {
|
||||
"requestedUnits": 13,
|
||||
"completedUnits": 2,
|
||||
"unsupportedUnits": 11,
|
||||
"unit": "CHECK"
|
||||
},
|
||||
"data": {
|
||||
"catalogVersion": 1,
|
||||
"resourceAlias": "cluster-1",
|
||||
"observations": {
|
||||
"capacity": {
|
||||
"source": "LIVE_DISK_PROBE",
|
||||
"observedAt": "2030-01-01T00:00:00Z",
|
||||
"freshness": "CURRENT",
|
||||
"complete": true,
|
||||
"totalBytes": 100,
|
||||
"usedBytes": 50
|
||||
},
|
||||
"coarseFlags": {
|
||||
"source": "LIVE_DISK_PROBE",
|
||||
"observedAt": "2030-01-01T00:00:00Z",
|
||||
"freshness": "CURRENT",
|
||||
"complete": true,
|
||||
"values": []
|
||||
}
|
||||
},
|
||||
"checks": [
|
||||
{
|
||||
"checkId": "capacity.exhausted",
|
||||
"outcome": "PASS",
|
||||
"reasonCode": "CAPACITY_REMAINING",
|
||||
"resourceAlias": "cluster-1"
|
||||
},
|
||||
{
|
||||
"checkId": "cluster.reportedFlags",
|
||||
"outcome": "PASS",
|
||||
"reasonCode": "NO_COARSE_CONDITION_REPORTED",
|
||||
"resourceAlias": "cluster-1"
|
||||
}
|
||||
],
|
||||
"unsupportedChecks": [
|
||||
{ "checkId": "memory.reportedPressure", "reasonCode": "EVIDENCE_CONTRACT_UNAVAILABLE" },
|
||||
{ "checkId": "tls.validity", "reasonCode": "EVIDENCE_CONTRACT_UNAVAILABLE" },
|
||||
{ "checkId": "cpu.sufficiency", "reasonCode": "EVIDENCE_CONTRACT_UNAVAILABLE" },
|
||||
{ "checkId": "swap.pressure", "reasonCode": "EVIDENCE_CONTRACT_UNAVAILABLE" },
|
||||
{ "checkId": "drive.health", "reasonCode": "EVIDENCE_CONTRACT_UNAVAILABLE" },
|
||||
{ "checkId": "inode.available", "reasonCode": "EVIDENCE_CONTRACT_UNAVAILABLE" },
|
||||
{ "checkId": "filesystem.compatibility", "reasonCode": "EVIDENCE_CONTRACT_UNAVAILABLE" },
|
||||
{ "checkId": "osKernel.compatibility", "reasonCode": "EVIDENCE_CONTRACT_UNAVAILABLE" },
|
||||
{ "checkId": "launch.permissions", "reasonCode": "EVIDENCE_CONTRACT_UNAVAILABLE" },
|
||||
{ "checkId": "version.consistency", "reasonCode": "EVIDENCE_CONTRACT_UNAVAILABLE" },
|
||||
{ "checkId": "kubernetes.configuration", "reasonCode": "EVIDENCE_CONTRACT_UNAVAILABLE" }
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
{
|
||||
"protocolVersion": "v1",
|
||||
"fixtureSet": "diagnostic-scheduler",
|
||||
"fixture": "health-service-reject-vectors",
|
||||
"baseFixture": "health-service-accept-vectors.json",
|
||||
"vectors": [
|
||||
{ "id": "health-service.job.old-node-capability", "base": "health-service.job.explicit-freshness", "path": "/requiredCapabilities/0", "replacement": "health.check@1", "reason": "UNSUPPORTED_TOOL" },
|
||||
{ "id": "health-service.job.missing-capability", "base": "health-service.job.explicit-freshness", "path": "/requiredCapabilities", "replacement": [], "reason": "UNSUPPORTED_TOOL" },
|
||||
{ "id": "health-service.job.missing-freshness-policy", "base": "health-service.job.explicit-freshness", "path": "/parameters/maxEvidenceAgeSeconds", "remove": true, "reason": "LIMIT_EXCEEDED" },
|
||||
{ "id": "health-service.job.profile-parameter", "base": "health-service.job.explicit-freshness", "path": "/parameters/durationMillis", "replacement": 1000, "reason": "INVALID" },
|
||||
{ "id": "health-service.job.oversized-result-budget", "base": "health-service.job.explicit-freshness", "path": "/limits/maxOutputBytes", "replacement": 262145, "reason": "LIMIT_EXCEEDED" },
|
||||
{ "id": "health-service.job.wrong-target", "base": "health-service.job.explicit-freshness", "path": "/deviceName", "replacement": "organizations/018cc251-f400-7abc-8def-0123456789ab/clusters/018cc251-f400-7abc-8def-0123456789ac/clusterDevices/018cc251-f400-7abc-8def-0123456789ae", "reason": "TARGET_MISMATCH" },
|
||||
{ "id": "health-service.job.wrong-nonce", "base": "health-service.job.explicit-freshness", "path": "/nonce", "replacement": "invalid", "reason": "INVALID" },
|
||||
{ "id": "health-service.job.wrong-signature", "base": "health-service.job.explicit-freshness", "path": "/signature/value", "replacement": "invalid", "reason": "SIGNATURE_INVALID" },
|
||||
{ "id": "health-service.job.revoked-consent", "base": "health-service.job.explicit-freshness", "path": "/parameters/consentExpiresAt", "replacement": "2029-12-31T23:59:59Z", "reason": "EXPIRED" },
|
||||
{ "id": "health-service.result.raw-admin-data", "base": "health-service.result.current-partial-catalog", "path": "/data/rawAdmin", "replacement": { "endpoint": "forbidden" }, "reason": "UNKNOWN_FIELD" },
|
||||
{ "id": "health-service.result.missing-unsupported", "base": "health-service.result.current-partial-catalog", "path": "/data/unsupportedChecks/10", "remove": true, "reason": "INCOMPLETE_COVERAGE" },
|
||||
{ "id": "health-service.result.path-leak", "base": "health-service.result.current-partial-catalog", "path": "/data/observations/capacity/path", "replacement": "/customer/data", "reason": "UNKNOWN_FIELD" }
|
||||
]
|
||||
}
|
||||
@@ -1,6 +1,6 @@
|
||||
975c1ca53eefeef6766a6fc0b3d3281f7408255342b0686e5e2aee5ad055414c duplicate.json
|
||||
963529a38a02849c6c2acc6d72668dca9f63218b49c89fae41a451b584850411 overflow.json
|
||||
60300d053dd3746a3aceb327315e8c5abe292fb6505dd02a64b8d6279fb56c09 producer-capabilities.json
|
||||
5d8ec114a5c3d9633d14ea3a227a0fdf6dfc04aca5de1fce39fce7d139b3dc08 producer-capabilities.json
|
||||
e3adeee1c8a19aa17e70894896fb79c072e3785bea3611b93c11e79f039ed5af stale.json
|
||||
22cc7337f545271ebf11ca6b76e7e479d62c51278ac001a8d870b07d0fa9884e unknown.json
|
||||
b06a72ff1d964efe65e996797705ec2ef72574a813aec5d8e9aff26c899f1352 valid.json
|
||||
|
||||
@@ -9,6 +9,7 @@
|
||||
"inventory.environment@1"
|
||||
],
|
||||
"producerCapabilities": [
|
||||
"health.check.service@1",
|
||||
"performance.client@1",
|
||||
"performance.drive@1",
|
||||
"performance.network@1",
|
||||
@@ -33,6 +34,7 @@
|
||||
"heartbeat",
|
||||
"diagnostics.policy.v1",
|
||||
"inventory.environment@1",
|
||||
"health.check.service@1",
|
||||
"performance.client@1",
|
||||
"performance.drive@1",
|
||||
"performance.network@1",
|
||||
|
||||
@@ -1,13 +1,13 @@
|
||||
a499742c06046e9cf2f781fc17d8b4cf95e018ab45fa41ee5944ba15b0e2902c auth
|
||||
9b54e39584a442270ae486e8378df1d6f168c45d0c59baba9459b46bb212d2df version
|
||||
159d116e965b51771f1687fc6ec9bdfbf2a4a0e6d6aa26d3f1046b3dac2697dd registration
|
||||
9b3862f5160968a2ff281afb3ea519a3d49ae1bc3d767153d70f309d2a7457a7 heartbeat
|
||||
8a92d233dd164f2648905595c9a31d533cc35b3bfcb4a79babd1f81a297332e6 heartbeat
|
||||
cc6c3f3bf5e6a938f2e00b8ce42d8fa7bfb974f4eb2927101d4ede4ebdc10dd5 inventory
|
||||
0207458ec9b97b368e2f347da112511a3aabd27d752890cacda4fa2408e7f19e object-performance
|
||||
a464d4c695d76dd985c0f2abf68c2135c080221098ad2427022f85243a7bf6ae network-performance
|
||||
813cbbf209e8fc9ddaafaeb4eb4745024b1c0edcb26634073a85023844141477 site-replication-performance
|
||||
fad21eec58d4d4d547893af9050cdc5eaf9d4f84b15fb423214b16be50e7ff7a telemetry
|
||||
484da2e7599148d36e1ac19d0ae5d305c4a96da7a00495a8ea202d3bae240a88 diagnostic-scheduler
|
||||
4f51f7916c8534ca6258785c0d90c83ea1465fba276ba180c73720ded2b71cd4 diagnostic-scheduler
|
||||
a2557ab1f7f70fb86a8affc8451596c272478fd2d5500c367b8f9e8385d72578 offline-enrollment
|
||||
8b6fd07759d1264489a8bb54d2dda4836ad0f59b5ee5d52966365918b3de4c9e bundle
|
||||
42cc936dc8fe87335ebff5abe01e3ebf85bdada7f37b43bfd8add46435c9d6f9 redaction
|
||||
|
||||
@@ -0,0 +1,237 @@
|
||||
{
|
||||
"$schema": "https://json-schema.org/draft/2020-12/schema",
|
||||
"$id": "https://connect.rustfs.com/schemas/agent/v1/health-service-result.schema.json",
|
||||
"title": "RustFS bounded health service result v1",
|
||||
"type": "object",
|
||||
"additionalProperties": false,
|
||||
"required": ["schemaVersion", "runUid", "toolId", "capability", "outcome", "reasonCode", "durationMillis", "provenance", "coverage", "data"],
|
||||
"properties": {
|
||||
"schemaVersion": { "const": 1 },
|
||||
"runUid": { "$ref": "#/$defs/uuid7" },
|
||||
"toolId": { "const": "health.check" },
|
||||
"capability": { "const": "health.check.service@1" },
|
||||
"outcome": { "const": "PARTIAL" },
|
||||
"reasonCode": {
|
||||
"enum": ["CATALOG_PARTIAL", "EVIDENCE_INCOMPLETE", "EVIDENCE_STALE", "EVIDENCE_FRESHNESS_UNKNOWN", "CLOCK_SKEW"]
|
||||
},
|
||||
"durationMillis": { "type": "integer", "minimum": 0, "maximum": 30000 },
|
||||
"provenance": { "$ref": "#/$defs/provenance" },
|
||||
"coverage": { "$ref": "#/$defs/coverage" },
|
||||
"data": { "$ref": "#/$defs/data" }
|
||||
},
|
||||
"$defs": {
|
||||
"uuid7": {
|
||||
"type": "string",
|
||||
"format": "uuid",
|
||||
"pattern": "^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$"
|
||||
},
|
||||
"provenance": {
|
||||
"type": "object",
|
||||
"additionalProperties": false,
|
||||
"required": ["repository", "sourceCommit", "executableSha256", "rustfsVersion", "osFamily", "architecture", "buildFeatures"],
|
||||
"properties": {
|
||||
"repository": { "const": "rustfs/rustfs" },
|
||||
"sourceCommit": { "type": "string", "pattern": "^[0-9a-f]{40}$" },
|
||||
"executableSha256": { "type": "string", "pattern": "^[0-9a-f]{64}$" },
|
||||
"rustfsVersion": { "type": "string", "minLength": 1, "maxLength": 64 },
|
||||
"osFamily": { "enum": ["LINUX", "DARWIN", "WINDOWS", "FREEBSD", "OTHER"] },
|
||||
"architecture": { "enum": ["x86_64", "aarch64", "other"] },
|
||||
"buildFeatures": {
|
||||
"type": "array",
|
||||
"items": { "type": "string", "pattern": "^[A-Za-z0-9._-]{1,64}$" },
|
||||
"maxItems": 64,
|
||||
"uniqueItems": true
|
||||
}
|
||||
}
|
||||
},
|
||||
"coverage": {
|
||||
"type": "object",
|
||||
"additionalProperties": false,
|
||||
"required": ["requestedUnits", "completedUnits", "unsupportedUnits", "unit"],
|
||||
"properties": {
|
||||
"requestedUnits": { "const": 13 },
|
||||
"completedUnits": { "const": 2 },
|
||||
"unsupportedUnits": { "const": 11 },
|
||||
"unit": { "const": "CHECK" }
|
||||
}
|
||||
},
|
||||
"data": {
|
||||
"type": "object",
|
||||
"additionalProperties": false,
|
||||
"required": ["catalogVersion", "resourceAlias", "observations", "checks", "unsupportedChecks"],
|
||||
"properties": {
|
||||
"catalogVersion": { "const": 1 },
|
||||
"resourceAlias": { "const": "cluster-1" },
|
||||
"observations": { "$ref": "#/$defs/observations" },
|
||||
"checks": {
|
||||
"type": "array",
|
||||
"prefixItems": [
|
||||
{ "$ref": "#/$defs/capacityCheck" },
|
||||
{ "$ref": "#/$defs/flagsCheck" }
|
||||
],
|
||||
"items": false,
|
||||
"minItems": 2,
|
||||
"maxItems": 2
|
||||
},
|
||||
"unsupportedChecks": {
|
||||
"type": "array",
|
||||
"prefixItems": [
|
||||
{ "$ref": "#/$defs/unsupportedMemory" },
|
||||
{ "$ref": "#/$defs/unsupportedTls" },
|
||||
{ "$ref": "#/$defs/unsupportedCpu" },
|
||||
{ "$ref": "#/$defs/unsupportedSwap" },
|
||||
{ "$ref": "#/$defs/unsupportedDrive" },
|
||||
{ "$ref": "#/$defs/unsupportedInode" },
|
||||
{ "$ref": "#/$defs/unsupportedFilesystem" },
|
||||
{ "$ref": "#/$defs/unsupportedOsKernel" },
|
||||
{ "$ref": "#/$defs/unsupportedLaunch" },
|
||||
{ "$ref": "#/$defs/unsupportedVersion" },
|
||||
{ "$ref": "#/$defs/unsupportedKubernetes" }
|
||||
],
|
||||
"items": false,
|
||||
"minItems": 11,
|
||||
"maxItems": 11
|
||||
}
|
||||
}
|
||||
},
|
||||
"observations": {
|
||||
"type": "object",
|
||||
"additionalProperties": false,
|
||||
"required": ["capacity", "coarseFlags"],
|
||||
"properties": {
|
||||
"capacity": { "$ref": "#/$defs/capacityObservation" },
|
||||
"coarseFlags": { "$ref": "#/$defs/flagsObservation" }
|
||||
}
|
||||
},
|
||||
"capacityObservation": {
|
||||
"type": "object",
|
||||
"additionalProperties": false,
|
||||
"required": ["source", "observedAt", "freshness", "complete", "totalBytes", "usedBytes"],
|
||||
"properties": {
|
||||
"source": { "const": "LIVE_DISK_PROBE" },
|
||||
"observedAt": { "type": ["string", "null"], "format": "date-time" },
|
||||
"freshness": { "$ref": "#/$defs/freshness" },
|
||||
"complete": { "type": "boolean" },
|
||||
"totalBytes": { "$ref": "#/$defs/nullableCounter" },
|
||||
"usedBytes": { "$ref": "#/$defs/nullableCounter" }
|
||||
},
|
||||
"allOf": [
|
||||
{
|
||||
"if": { "properties": { "complete": { "const": true } } },
|
||||
"then": {
|
||||
"properties": {
|
||||
"observedAt": { "type": "string", "format": "date-time" },
|
||||
"totalBytes": { "type": "integer", "minimum": 0, "maximum": 9007199254740991 },
|
||||
"usedBytes": { "type": "integer", "minimum": 0, "maximum": 9007199254740991 }
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
"flagsObservation": {
|
||||
"type": "object",
|
||||
"additionalProperties": false,
|
||||
"required": ["source", "observedAt", "freshness", "complete", "values"],
|
||||
"properties": {
|
||||
"source": { "const": "LIVE_DISK_PROBE" },
|
||||
"observedAt": { "type": ["string", "null"], "format": "date-time" },
|
||||
"freshness": { "$ref": "#/$defs/freshness" },
|
||||
"complete": { "type": "boolean" },
|
||||
"values": {
|
||||
"oneOf": [
|
||||
{ "type": "null" },
|
||||
{
|
||||
"type": "array",
|
||||
"items": {
|
||||
"enum": ["capacity.critical", "capacity.warning", "clock.skew", "cluster.degraded", "cluster.healing", "cluster.readonly", "drive.offline", "node.offline"]
|
||||
},
|
||||
"maxItems": 8,
|
||||
"uniqueItems": true
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
"allOf": [
|
||||
{
|
||||
"if": { "properties": { "complete": { "const": true } } },
|
||||
"then": {
|
||||
"properties": {
|
||||
"observedAt": { "type": "string", "format": "date-time" },
|
||||
"values": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"enum": ["capacity.critical", "capacity.warning", "clock.skew", "cluster.degraded", "cluster.healing", "cluster.readonly", "drive.offline", "node.offline"]
|
||||
},
|
||||
"maxItems": 8,
|
||||
"uniqueItems": true
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
"freshness": { "enum": ["CURRENT", "STALE", "UNKNOWN"] },
|
||||
"nullableCounter": {
|
||||
"type": ["integer", "null"],
|
||||
"minimum": 0,
|
||||
"maximum": 9007199254740991
|
||||
},
|
||||
"capacityCheck": {
|
||||
"allOf": [
|
||||
{ "$ref": "#/$defs/check" },
|
||||
{
|
||||
"properties": {
|
||||
"checkId": { "const": "capacity.exhausted" },
|
||||
"reasonCode": {
|
||||
"enum": ["CAPACITY_EXHAUSTED", "CAPACITY_REMAINING", "CAPACITY_UNAVAILABLE", "EVIDENCE_MISSING", "INVALID_EVIDENCE", "EVIDENCE_STALE", "EVIDENCE_FRESHNESS_UNKNOWN", "CLOCK_SKEW"]
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
"flagsCheck": {
|
||||
"allOf": [
|
||||
{ "$ref": "#/$defs/check" },
|
||||
{
|
||||
"properties": {
|
||||
"checkId": { "const": "cluster.reportedFlags" },
|
||||
"reasonCode": {
|
||||
"enum": ["COARSE_CONDITION_REPORTED", "NO_COARSE_CONDITION_REPORTED", "EVIDENCE_MISSING", "INVALID_EVIDENCE", "EVIDENCE_STALE", "EVIDENCE_FRESHNESS_UNKNOWN", "CLOCK_SKEW"]
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
"check": {
|
||||
"type": "object",
|
||||
"additionalProperties": false,
|
||||
"required": ["checkId", "outcome", "reasonCode", "resourceAlias"],
|
||||
"properties": {
|
||||
"checkId": { "type": "string" },
|
||||
"outcome": { "enum": ["PASS", "FAIL", "UNKNOWN"] },
|
||||
"reasonCode": { "type": "string" },
|
||||
"resourceAlias": { "const": "cluster-1" }
|
||||
}
|
||||
},
|
||||
"unsupported": {
|
||||
"type": "object",
|
||||
"additionalProperties": false,
|
||||
"required": ["checkId", "reasonCode"],
|
||||
"properties": {
|
||||
"checkId": { "type": "string" },
|
||||
"reasonCode": { "const": "EVIDENCE_CONTRACT_UNAVAILABLE" }
|
||||
}
|
||||
},
|
||||
"unsupportedMemory": { "allOf": [{ "$ref": "#/$defs/unsupported" }, { "properties": { "checkId": { "const": "memory.reportedPressure" } } }] },
|
||||
"unsupportedTls": { "allOf": [{ "$ref": "#/$defs/unsupported" }, { "properties": { "checkId": { "const": "tls.validity" } } }] },
|
||||
"unsupportedCpu": { "allOf": [{ "$ref": "#/$defs/unsupported" }, { "properties": { "checkId": { "const": "cpu.sufficiency" } } }] },
|
||||
"unsupportedSwap": { "allOf": [{ "$ref": "#/$defs/unsupported" }, { "properties": { "checkId": { "const": "swap.pressure" } } }] },
|
||||
"unsupportedDrive": { "allOf": [{ "$ref": "#/$defs/unsupported" }, { "properties": { "checkId": { "const": "drive.health" } } }] },
|
||||
"unsupportedInode": { "allOf": [{ "$ref": "#/$defs/unsupported" }, { "properties": { "checkId": { "const": "inode.available" } } }] },
|
||||
"unsupportedFilesystem": { "allOf": [{ "$ref": "#/$defs/unsupported" }, { "properties": { "checkId": { "const": "filesystem.compatibility" } } }] },
|
||||
"unsupportedOsKernel": { "allOf": [{ "$ref": "#/$defs/unsupported" }, { "properties": { "checkId": { "const": "osKernel.compatibility" } } }] },
|
||||
"unsupportedLaunch": { "allOf": [{ "$ref": "#/$defs/unsupported" }, { "properties": { "checkId": { "const": "launch.permissions" } } }] },
|
||||
"unsupportedVersion": { "allOf": [{ "$ref": "#/$defs/unsupported" }, { "properties": { "checkId": { "const": "version.consistency" } } }] },
|
||||
"unsupportedKubernetes": { "allOf": [{ "$ref": "#/$defs/unsupported" }, { "properties": { "checkId": { "const": "kubernetes.configuration" } } }] }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,362 @@
|
||||
{
|
||||
"$schema": "https://json-schema.org/draft/2020-12/schema",
|
||||
"$id": "https://connect.rustfs.com/schemas/agent/v1/job-envelope.schema.json",
|
||||
"title": "RustFS Connect Agent Job Envelope v1",
|
||||
"type": "object",
|
||||
"additionalProperties": false,
|
||||
"required": [
|
||||
"jobId",
|
||||
"protocolVersion",
|
||||
"jobType",
|
||||
"schemaVersion",
|
||||
"organizationName",
|
||||
"clusterName",
|
||||
"deviceName",
|
||||
"createTime",
|
||||
"expireTime",
|
||||
"nonce",
|
||||
"requiredCapabilities",
|
||||
"authorization",
|
||||
"limits",
|
||||
"parameters",
|
||||
"signature"
|
||||
],
|
||||
"properties": {
|
||||
"jobId": { "type": "string", "format": "uuid", "pattern": "^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$" },
|
||||
"protocolVersion": { "const": "v1" },
|
||||
"jobType": { "enum": ["health.check", "profile.cpu", "profile.memory", "profile.threads", "performance.drive", "performance.network", "top.api", "top.locks", "top.rpc"] },
|
||||
"schemaVersion": { "const": 1 },
|
||||
"organizationName": { "type": "string", "pattern": "^organizations/[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$" },
|
||||
"clusterName": { "type": "string", "pattern": "^organizations/[0-9a-f-]{36}/clusters/[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$" },
|
||||
"deviceName": { "type": "string", "pattern": "^organizations/[0-9a-f-]{36}/clusters/[0-9a-f-]{36}/clusterDevices/[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$" },
|
||||
"createTime": { "type": "string", "format": "date-time" },
|
||||
"expireTime": { "type": "string", "format": "date-time" },
|
||||
"nonce": { "type": "string", "pattern": "^[A-Za-z0-9_-]{43}$" },
|
||||
"requiredCapabilities": {
|
||||
"type": "array",
|
||||
"prefixItems": [{ "enum": ["health.check.service@1", "profile.cpu@1", "profile.memory.service@1", "profile.threads@1", "performance.drive@1", "performance.network@1", "top.api@1", "top.locks@1", "top.rpc@1"] }],
|
||||
"items": false,
|
||||
"minItems": 1,
|
||||
"maxItems": 1,
|
||||
"uniqueItems": true
|
||||
},
|
||||
"limits": {
|
||||
"type": "object",
|
||||
"additionalProperties": false,
|
||||
"required": ["timeoutSeconds", "maxOutputBytes", "maxMemoryBytes", "maxCpuMillis"],
|
||||
"properties": {
|
||||
"timeoutSeconds": { "type": "integer", "minimum": 1, "maximum": 30 },
|
||||
"maxOutputBytes": { "type": "integer", "minimum": 1, "maximum": 524288 },
|
||||
"maxMemoryBytes": { "type": "integer", "minimum": 1, "maximum": 67108864 },
|
||||
"maxCpuMillis": { "type": "integer", "minimum": 1, "maximum": 30000 }
|
||||
}
|
||||
},
|
||||
"authorization": {
|
||||
"type": "object",
|
||||
"additionalProperties": false,
|
||||
"required": ["actorType", "actorName", "requestId"],
|
||||
"properties": {
|
||||
"actorType": { "const": "BROWSER_USER" },
|
||||
"actorName": { "type": "string", "pattern": "^users/[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$" },
|
||||
"requestId": { "type": "string", "format": "uuid", "pattern": "^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$" }
|
||||
}
|
||||
},
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"additionalProperties": false,
|
||||
"required": ["artifactUid", "consentUid", "consentPolicyRevision", "consentExpiresAt"],
|
||||
"properties": {
|
||||
"artifactUid": { "type": "string", "format": "uuid", "pattern": "^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$" },
|
||||
"consentUid": { "type": "string", "format": "uuid", "pattern": "^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$" },
|
||||
"consentPolicyRevision": { "type": "integer", "minimum": 1 },
|
||||
"consentExpiresAt": { "type": "string", "format": "date-time" },
|
||||
"durationMillis": { "type": "integer", "minimum": 1, "maximum": 30000 },
|
||||
"samplePeriodMicros": { "type": "integer", "minimum": 1, "maximum": 30000000 },
|
||||
"maxEvidenceAgeSeconds": { "type": "integer", "minimum": 1, "maximum": 86400 },
|
||||
"targetAlias": { "const": "drive-1" },
|
||||
"scratchBytes": { "const": 524288 },
|
||||
"blockBytes": { "const": 65536 },
|
||||
"trafficBytes": { "type": "integer", "minimum": 1, "maximum": 1048576 }
|
||||
}
|
||||
},
|
||||
"signature": {
|
||||
"type": "object",
|
||||
"additionalProperties": false,
|
||||
"required": ["algorithm", "keyId", "value"],
|
||||
"properties": {
|
||||
"algorithm": { "const": "Ed25519" },
|
||||
"keyId": { "type": "string", "minLength": 1 },
|
||||
"value": { "type": "string", "minLength": 43 }
|
||||
}
|
||||
}
|
||||
},
|
||||
"allOf": [
|
||||
{
|
||||
"if": { "properties": { "jobType": { "const": "health.check" } } },
|
||||
"then": {
|
||||
"properties": {
|
||||
"requiredCapabilities": {
|
||||
"type": "array",
|
||||
"prefixItems": [{ "const": "health.check.service@1" }],
|
||||
"items": false,
|
||||
"minItems": 1,
|
||||
"maxItems": 1
|
||||
},
|
||||
"limits": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"timeoutSeconds": { "const": 30 },
|
||||
"maxOutputBytes": { "const": 262144 },
|
||||
"maxMemoryBytes": { "const": 67108864 },
|
||||
"maxCpuMillis": { "const": 5000 }
|
||||
}
|
||||
},
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"required": ["maxEvidenceAgeSeconds"],
|
||||
"not": {
|
||||
"anyOf": [
|
||||
{ "required": ["durationMillis"] },
|
||||
{ "required": ["samplePeriodMicros"] },
|
||||
{ "required": ["targetAlias"] },
|
||||
{ "required": ["scratchBytes"] },
|
||||
{ "required": ["blockBytes"] },
|
||||
{ "required": ["trafficBytes"] }
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"else": {
|
||||
"properties": {
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"required": ["durationMillis", "samplePeriodMicros"],
|
||||
"not": { "required": ["maxEvidenceAgeSeconds"] }
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"if": { "properties": { "jobType": { "const": "performance.drive" } } },
|
||||
"then": {
|
||||
"properties": {
|
||||
"requiredCapabilities": {
|
||||
"type": "array",
|
||||
"prefixItems": [{ "const": "performance.drive@1" }],
|
||||
"items": false,
|
||||
"minItems": 1,
|
||||
"maxItems": 1
|
||||
},
|
||||
"limits": {
|
||||
"type": "object",
|
||||
"properties": { "maxCpuMillis": { "type": "integer", "maximum": 5000 } }
|
||||
},
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"required": ["targetAlias", "scratchBytes", "blockBytes"],
|
||||
"properties": {
|
||||
"durationMillis": { "const": 5000 },
|
||||
"targetAlias": { "const": "drive-1" },
|
||||
"scratchBytes": { "const": 524288 },
|
||||
"blockBytes": { "const": 65536 }
|
||||
},
|
||||
"not": {
|
||||
"type": "object",
|
||||
"properties": { "trafficBytes": true },
|
||||
"required": ["trafficBytes"]
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"else": {
|
||||
"properties": {
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"not": {
|
||||
"anyOf": [
|
||||
{ "required": ["targetAlias"] },
|
||||
{ "required": ["scratchBytes"] },
|
||||
{ "required": ["blockBytes"] }
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"if": { "properties": { "jobType": { "const": "performance.network" } } },
|
||||
"then": {
|
||||
"properties": {
|
||||
"requiredCapabilities": {
|
||||
"type": "array",
|
||||
"prefixItems": [{ "const": "performance.network@1" }],
|
||||
"items": false,
|
||||
"minItems": 1,
|
||||
"maxItems": 1
|
||||
},
|
||||
"limits": {
|
||||
"type": "object",
|
||||
"properties": { "maxCpuMillis": { "type": "integer", "maximum": 5000 } }
|
||||
},
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"required": ["trafficBytes"],
|
||||
"properties": { "trafficBytes": { "type": "integer", "minimum": 1, "maximum": 1048576 } }
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"if": { "properties": { "jobType": { "const": "profile.cpu" } } },
|
||||
"then": {
|
||||
"properties": {
|
||||
"requiredCapabilities": {
|
||||
"type": "array",
|
||||
"prefixItems": [{ "const": "profile.cpu@1" }],
|
||||
"items": false,
|
||||
"minItems": 1,
|
||||
"maxItems": 1
|
||||
},
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"not": {
|
||||
"type": "object",
|
||||
"properties": { "trafficBytes": true },
|
||||
"required": ["trafficBytes"]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"if": { "properties": { "jobType": { "const": "profile.memory" } } },
|
||||
"then": {
|
||||
"properties": {
|
||||
"requiredCapabilities": {
|
||||
"type": "array",
|
||||
"prefixItems": [{ "const": "profile.memory.service@1" }],
|
||||
"items": false,
|
||||
"minItems": 1,
|
||||
"maxItems": 1
|
||||
},
|
||||
"limits": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"maxCpuMillis": { "type": "integer", "maximum": 5000 },
|
||||
"maxMemoryBytes": { "const": 67108864 }
|
||||
}
|
||||
},
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"not": {
|
||||
"type": "object",
|
||||
"properties": { "trafficBytes": true },
|
||||
"required": ["trafficBytes"]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"if": { "properties": { "jobType": { "const": "profile.threads" } } },
|
||||
"then": {
|
||||
"properties": {
|
||||
"requiredCapabilities": {
|
||||
"type": "array",
|
||||
"prefixItems": [{ "const": "profile.threads@1" }],
|
||||
"items": false,
|
||||
"minItems": 1,
|
||||
"maxItems": 1
|
||||
},
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"not": {
|
||||
"type": "object",
|
||||
"properties": { "trafficBytes": true },
|
||||
"required": ["trafficBytes"]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"if": { "properties": { "jobType": { "const": "top.api" } } },
|
||||
"then": {
|
||||
"properties": {
|
||||
"requiredCapabilities": {
|
||||
"type": "array",
|
||||
"prefixItems": [{ "const": "top.api@1" }],
|
||||
"items": false,
|
||||
"minItems": 1,
|
||||
"maxItems": 1
|
||||
},
|
||||
"limits": {
|
||||
"type": "object",
|
||||
"properties": { "maxCpuMillis": { "type": "integer", "maximum": 5000 } }
|
||||
},
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"not": {
|
||||
"type": "object",
|
||||
"properties": { "trafficBytes": true },
|
||||
"required": ["trafficBytes"]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"if": { "properties": { "jobType": { "const": "top.rpc" } } },
|
||||
"then": {
|
||||
"properties": {
|
||||
"requiredCapabilities": {
|
||||
"type": "array",
|
||||
"prefixItems": [{ "const": "top.rpc@1" }],
|
||||
"items": false,
|
||||
"minItems": 1,
|
||||
"maxItems": 1
|
||||
},
|
||||
"limits": {
|
||||
"type": "object",
|
||||
"properties": { "maxCpuMillis": { "type": "integer", "maximum": 5000 } }
|
||||
},
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"not": {
|
||||
"type": "object",
|
||||
"properties": { "trafficBytes": true },
|
||||
"required": ["trafficBytes"]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
"if": { "properties": { "jobType": { "const": "top.locks" } } },
|
||||
"then": {
|
||||
"properties": {
|
||||
"requiredCapabilities": {
|
||||
"type": "array",
|
||||
"prefixItems": [{ "const": "top.locks@1" }],
|
||||
"items": false,
|
||||
"minItems": 1,
|
||||
"maxItems": 1
|
||||
},
|
||||
"limits": {
|
||||
"type": "object",
|
||||
"properties": { "maxCpuMillis": { "type": "integer", "maximum": 5000 } }
|
||||
},
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"not": {
|
||||
"type": "object",
|
||||
"properties": { "trafficBytes": true },
|
||||
"required": ["trafficBytes"]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,827 @@
|
||||
// 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.
|
||||
|
||||
//! Bounded online HealthCheck observation and signed export.
|
||||
//!
|
||||
//! Only the approved L0 cluster aggregate is exported: capacity counters and
|
||||
//! the existing coarse flag allow-list. Per-drive identity, endpoints, paths,
|
||||
//! raw admin responses and object data never enter the result.
|
||||
|
||||
use std::collections::BTreeSet;
|
||||
use std::io::{Cursor, Write as _};
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::time::Instant;
|
||||
|
||||
use base64_simd::URL_SAFE_NO_PAD;
|
||||
use chrono::{DateTime, SecondsFormat, Utc};
|
||||
use p256::ecdsa::{Signature, SigningKey, signature::Signer as _};
|
||||
use p256::pkcs8::DecodePrivateKey as _;
|
||||
use serde::Serialize;
|
||||
use sha2::{Digest as _, Sha256};
|
||||
use thiserror::Error;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use uuid::{Uuid, Variant, Version};
|
||||
use zip::{CompressionMethod, ZipWriter, write::SimpleFileOptions};
|
||||
|
||||
use super::ProfileProvenance;
|
||||
use crate::connect::DeviceIdentity;
|
||||
use crate::storage::storage_api::contract::admin::{DiskSetSelector, StorageAdminApi};
|
||||
use crate::storage::storage_api::{DiskInfoOptions, StorageDiskRpcExt};
|
||||
|
||||
pub const HEALTH_SCHEMA_VERSION: u16 = 1;
|
||||
pub const HEALTH_TOOL_ID: &str = "health.check";
|
||||
pub const HEALTH_SERVICE_CAPABILITY: &str = "health.check.service@1";
|
||||
pub const HEALTH_TIMEOUT_SECONDS: u64 = 30;
|
||||
pub const MAX_HEALTH_OUTPUT_BYTES: u64 = 262_144;
|
||||
pub const MAX_HEALTH_MEMORY_BYTES: u64 = 67_108_864;
|
||||
pub const MAX_HEALTH_CPU_MILLIS: u64 = 5_000;
|
||||
pub const MAX_EVIDENCE_AGE_SECONDS: u64 = 86_400;
|
||||
pub const HEALTH_CATALOG_CHECKS: usize = 13;
|
||||
|
||||
const MAX_HEALTH_DRIVES: usize = 1_024;
|
||||
const MAX_SAFE_INTEGER: u64 = 9_007_199_254_740_991;
|
||||
const MAX_ENVELOPE_BYTES: usize = 16_384;
|
||||
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 RESOURCE_ALIAS: &str = "cluster-1";
|
||||
|
||||
const ALLOWED_FLAGS: [&str; 8] = [
|
||||
"capacity.critical",
|
||||
"capacity.warning",
|
||||
"clock.skew",
|
||||
"cluster.degraded",
|
||||
"cluster.healing",
|
||||
"cluster.readonly",
|
||||
"drive.offline",
|
||||
"node.offline",
|
||||
];
|
||||
|
||||
const UNSUPPORTED_CHECKS: [&str; 11] = [
|
||||
"memory.reportedPressure",
|
||||
"tls.validity",
|
||||
"cpu.sufficiency",
|
||||
"swap.pressure",
|
||||
"drive.health",
|
||||
"inode.available",
|
||||
"filesystem.compatibility",
|
||||
"osKernel.compatibility",
|
||||
"launch.permissions",
|
||||
"version.consistency",
|
||||
"kubernetes.configuration",
|
||||
];
|
||||
|
||||
static HEALTH_COLLECTOR_ACTIVE: AtomicBool = AtomicBool::new(false);
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct LocalHealthConsent {
|
||||
pub consent_uid: String,
|
||||
pub policy_revision: u64,
|
||||
pub expires_at_unix: i64,
|
||||
pub active: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct HealthServiceRequest {
|
||||
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: LocalHealthConsent,
|
||||
pub produced_at_unix: i64,
|
||||
pub expires_at_unix: i64,
|
||||
pub nonce: [u8; 32],
|
||||
pub max_evidence_age_seconds: u64,
|
||||
pub provenance: ProfileProvenance,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct HealthSourceObservation {
|
||||
pub observed_at_unix: Option<i64>,
|
||||
pub capacity_complete: bool,
|
||||
pub total_bytes: Option<u64>,
|
||||
pub used_bytes: Option<u64>,
|
||||
pub coarse_flags_complete: bool,
|
||||
pub coarse_flags: Option<Vec<String>>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
|
||||
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
|
||||
pub enum HealthOutcome {
|
||||
Partial,
|
||||
}
|
||||
|
||||
impl HealthOutcome {
|
||||
pub const fn as_str(self) -> &'static str {
|
||||
"PARTIAL"
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
|
||||
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
|
||||
pub enum HealthResultReason {
|
||||
CatalogPartial,
|
||||
EvidenceIncomplete,
|
||||
EvidenceStale,
|
||||
EvidenceFreshnessUnknown,
|
||||
ClockSkew,
|
||||
}
|
||||
|
||||
impl HealthResultReason {
|
||||
pub const fn as_str(self) -> &'static str {
|
||||
match self {
|
||||
Self::CatalogPartial => "CATALOG_PARTIAL",
|
||||
Self::EvidenceIncomplete => "EVIDENCE_INCOMPLETE",
|
||||
Self::EvidenceStale => "EVIDENCE_STALE",
|
||||
Self::EvidenceFreshnessUnknown => "EVIDENCE_FRESHNESS_UNKNOWN",
|
||||
Self::ClockSkew => "CLOCK_SKEW",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
|
||||
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
|
||||
pub enum HealthRuleOutcome {
|
||||
Pass,
|
||||
Fail,
|
||||
Unknown,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
|
||||
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
|
||||
pub enum HealthFreshness {
|
||||
Current,
|
||||
Stale,
|
||||
Unknown,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct HealthCheckResult {
|
||||
pub check_id: &'static str,
|
||||
pub outcome: HealthRuleOutcome,
|
||||
pub reason_code: &'static str,
|
||||
pub resource_alias: &'static str,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct UnsupportedHealthCheck {
|
||||
pub check_id: &'static str,
|
||||
pub reason_code: &'static str,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct HealthCapacityObservation {
|
||||
source: &'static str,
|
||||
observed_at: Option<String>,
|
||||
freshness: HealthFreshness,
|
||||
complete: bool,
|
||||
total_bytes: Option<u64>,
|
||||
used_bytes: Option<u64>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct HealthFlagsObservation {
|
||||
source: &'static str,
|
||||
observed_at: Option<String>,
|
||||
freshness: HealthFreshness,
|
||||
complete: bool,
|
||||
values: Option<Vec<String>>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct HealthObservations {
|
||||
capacity: HealthCapacityObservation,
|
||||
coarse_flags: HealthFlagsObservation,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct HealthData {
|
||||
catalog_version: u16,
|
||||
resource_alias: &'static str,
|
||||
observations: HealthObservations,
|
||||
checks: Vec<HealthCheckResult>,
|
||||
unsupported_checks: Vec<UnsupportedHealthCheck>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct HealthCoverage {
|
||||
requested_units: u32,
|
||||
completed_units: u32,
|
||||
unsupported_units: u32,
|
||||
unit: &'static str,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct HealthDiagnosticResult {
|
||||
schema_version: u16,
|
||||
run_uid: String,
|
||||
tool_id: &'static str,
|
||||
capability: &'static str,
|
||||
outcome: HealthOutcome,
|
||||
reason_code: HealthResultReason,
|
||||
duration_millis: u64,
|
||||
provenance: ProfileProvenance,
|
||||
coverage: HealthCoverage,
|
||||
data: HealthData,
|
||||
}
|
||||
|
||||
impl HealthDiagnosticResult {
|
||||
pub const fn outcome(&self) -> HealthOutcome {
|
||||
self.outcome
|
||||
}
|
||||
|
||||
pub const fn reason_code(&self) -> HealthResultReason {
|
||||
self.reason_code
|
||||
}
|
||||
|
||||
pub fn data(&self) -> &HealthData {
|
||||
&self.data
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct SignedHealthExport {
|
||||
pub artifact_uid: String,
|
||||
pub outcome: HealthOutcome,
|
||||
pub reason_code: HealthResultReason,
|
||||
pub envelope_json: Vec<u8>,
|
||||
pub envelope_signature: Vec<u8>,
|
||||
pub result_json: Vec<u8>,
|
||||
pub archive_bytes: Vec<u8>,
|
||||
pub archive_sha256: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Error, PartialEq, Eq)]
|
||||
pub enum HealthError {
|
||||
#[error("health_consent_required")]
|
||||
ConsentRequired,
|
||||
#[error("health_consent_expired")]
|
||||
ConsentExpired,
|
||||
#[error("health_request_expired")]
|
||||
Expired,
|
||||
#[error("health_invalid_request")]
|
||||
InvalidRequest,
|
||||
#[error("health_unsupported_contract")]
|
||||
Unsupported,
|
||||
#[error("health_limit_exceeded")]
|
||||
LimitExceeded,
|
||||
#[error("health_collection_cancelled")]
|
||||
Cancelled,
|
||||
#[error("health_collection_already_running")]
|
||||
Busy,
|
||||
#[error("health_source_unavailable")]
|
||||
SourceUnavailable,
|
||||
#[error("health_collection_failed")]
|
||||
CollectionFailed,
|
||||
#[error("health_export_signing_failed")]
|
||||
Signing,
|
||||
#[error("health_export_encoding_failed")]
|
||||
Encoding,
|
||||
}
|
||||
|
||||
struct CollectorLease;
|
||||
|
||||
impl CollectorLease {
|
||||
fn acquire() -> Result<Self, HealthError> {
|
||||
HEALTH_COLLECTOR_ACTIVE
|
||||
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
|
||||
.map(|_| Self)
|
||||
.map_err(|_| HealthError::Busy)
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for CollectorLease {
|
||||
fn drop(&mut self) {
|
||||
HEALTH_COLLECTOR_ACTIVE.store(false, Ordering::Release);
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn collect_runtime_health(
|
||||
request: &HealthServiceRequest,
|
||||
identity: &DeviceIdentity,
|
||||
cancel: &CancellationToken,
|
||||
) -> Result<SignedHealthExport, HealthError> {
|
||||
validate_request(request, Utc::now().timestamp())?;
|
||||
check_cancel(cancel)?;
|
||||
let _lease = CollectorLease::acquire()?;
|
||||
let started = Instant::now();
|
||||
let observation = collect_runtime_observation(cancel).await?;
|
||||
check_cancel(cancel)?;
|
||||
let result = evaluate_health_observation(request, observation, Utc::now(), started.elapsed());
|
||||
sign_health_export(request, &result, identity, cancel)
|
||||
}
|
||||
|
||||
pub fn evaluate_health_observation(
|
||||
request: &HealthServiceRequest,
|
||||
observation: HealthSourceObservation,
|
||||
evaluated_at: DateTime<Utc>,
|
||||
elapsed: std::time::Duration,
|
||||
) -> HealthDiagnosticResult {
|
||||
let observed_at = observation.observed_at_unix.and_then(DateTime::<Utc>::from_timestamp_secs);
|
||||
let freshness = freshness(observed_at, evaluated_at, request.max_evidence_age_seconds);
|
||||
let observed_at_text = observed_at.map(|value| value.to_rfc3339_opts(SecondsFormat::Secs, true));
|
||||
|
||||
let capacity = capacity_check(&observation, freshness);
|
||||
let flags = flags_check(&observation, freshness);
|
||||
let result_reason = result_reason(freshness, &observation);
|
||||
let unsupported_checks = UNSUPPORTED_CHECKS
|
||||
.into_iter()
|
||||
.map(|check_id| UnsupportedHealthCheck {
|
||||
check_id,
|
||||
reason_code: "EVIDENCE_CONTRACT_UNAVAILABLE",
|
||||
})
|
||||
.collect();
|
||||
|
||||
HealthDiagnosticResult {
|
||||
schema_version: HEALTH_SCHEMA_VERSION,
|
||||
run_uid: request.run_uid.clone(),
|
||||
tool_id: HEALTH_TOOL_ID,
|
||||
capability: HEALTH_SERVICE_CAPABILITY,
|
||||
outcome: HealthOutcome::Partial,
|
||||
reason_code: result_reason,
|
||||
duration_millis: u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX),
|
||||
provenance: request.provenance.clone(),
|
||||
coverage: HealthCoverage {
|
||||
requested_units: u32::try_from(HEALTH_CATALOG_CHECKS).unwrap_or(u32::MAX),
|
||||
completed_units: 2,
|
||||
unsupported_units: u32::try_from(UNSUPPORTED_CHECKS.len()).unwrap_or(u32::MAX),
|
||||
unit: "CHECK",
|
||||
},
|
||||
data: HealthData {
|
||||
catalog_version: 1,
|
||||
resource_alias: RESOURCE_ALIAS,
|
||||
observations: HealthObservations {
|
||||
capacity: HealthCapacityObservation {
|
||||
source: "LIVE_DISK_PROBE",
|
||||
observed_at: observed_at_text.clone(),
|
||||
freshness,
|
||||
complete: observation.capacity_complete,
|
||||
total_bytes: observation.total_bytes,
|
||||
used_bytes: observation.used_bytes,
|
||||
},
|
||||
coarse_flags: HealthFlagsObservation {
|
||||
source: "LIVE_DISK_PROBE",
|
||||
observed_at: observed_at_text,
|
||||
freshness,
|
||||
complete: observation.coarse_flags_complete,
|
||||
values: observation.coarse_flags,
|
||||
},
|
||||
},
|
||||
checks: vec![capacity, flags],
|
||||
unsupported_checks,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
async fn collect_runtime_observation(cancel: &CancellationToken) -> Result<HealthSourceObservation, HealthError> {
|
||||
let store = crate::runtime_sources::current_object_store_handle().ok_or(HealthError::SourceUnavailable)?;
|
||||
let backend = tokio::select! {
|
||||
_ = cancel.cancelled() => return Err(HealthError::Cancelled),
|
||||
backend = StorageAdminApi::backend_info(store.as_ref()) => backend,
|
||||
};
|
||||
let drive_counts = StorageAdminApi::set_drive_counts(store.as_ref());
|
||||
if backend.total_sets.is_empty()
|
||||
|| backend.total_sets.len() > MAX_HEALTH_DRIVES
|
||||
|| backend.total_sets.len() != drive_counts.len()
|
||||
|| drive_counts != backend.drives_per_set
|
||||
|| backend
|
||||
.total_sets
|
||||
.iter()
|
||||
.zip(&drive_counts)
|
||||
.any(|(&sets, &drives)| sets == 0 || drives == 0)
|
||||
{
|
||||
return Err(HealthError::SourceUnavailable);
|
||||
}
|
||||
let data_widths = if backend.standard_sc_data.is_empty() {
|
||||
None
|
||||
} else if backend.standard_sc_data.len() == drive_counts.len()
|
||||
&& backend
|
||||
.standard_sc_data
|
||||
.iter()
|
||||
.zip(&drive_counts)
|
||||
.all(|(&data, &drives)| (1..=drives).contains(&data))
|
||||
{
|
||||
Some(backend.standard_sc_data.as_slice())
|
||||
} else {
|
||||
return Err(HealthError::SourceUnavailable);
|
||||
};
|
||||
let expected = backend
|
||||
.total_sets
|
||||
.iter()
|
||||
.zip(&drive_counts)
|
||||
.try_fold(0usize, |total, (&sets, &drives)| {
|
||||
sets.checked_mul(drives).and_then(|count| total.checked_add(count))
|
||||
})
|
||||
.filter(|count| (1..=MAX_HEALTH_DRIVES).contains(count))
|
||||
.ok_or(HealthError::LimitExceeded)?;
|
||||
|
||||
let mut disks = Vec::with_capacity(expected);
|
||||
for (pool_idx, (&sets, &drives)) in backend.total_sets.iter().zip(&drive_counts).enumerate() {
|
||||
for set_idx in 0..sets {
|
||||
let set = tokio::select! {
|
||||
_ = cancel.cancelled() => return Err(HealthError::Cancelled),
|
||||
result = StorageAdminApi::disk_set_inventory(store.as_ref(), DiskSetSelector::new(pool_idx, set_idx)) => {
|
||||
result.map_err(|_| HealthError::SourceUnavailable)?
|
||||
}
|
||||
};
|
||||
if set.len() != drives || disks.len().saturating_add(set.len()) > expected {
|
||||
return Err(HealthError::SourceUnavailable);
|
||||
}
|
||||
disks.extend(
|
||||
set.into_iter()
|
||||
.enumerate()
|
||||
.map(|(disk_idx, disk)| (data_widths.is_none_or(|widths| disk_idx < widths[pool_idx]), disk)),
|
||||
);
|
||||
}
|
||||
}
|
||||
if disks.len() != expected {
|
||||
return Err(HealthError::SourceUnavailable);
|
||||
}
|
||||
|
||||
let observed_at_unix = Utc::now().timestamp();
|
||||
let mut total_bytes = 0u64;
|
||||
let mut used_bytes = 0u64;
|
||||
let mut capacity_complete = true;
|
||||
let mut flags_complete = true;
|
||||
let mut flags = BTreeSet::new();
|
||||
|
||||
for (include_capacity, disk) in disks {
|
||||
check_cancel(cancel)?;
|
||||
let Some(disk) = disk else {
|
||||
capacity_complete = false;
|
||||
flags.insert("cluster.degraded".to_owned());
|
||||
flags.insert("drive.offline".to_owned());
|
||||
continue;
|
||||
};
|
||||
let options = DiskInfoOptions {
|
||||
fresh_capacity: true,
|
||||
..Default::default()
|
||||
};
|
||||
let info = tokio::select! {
|
||||
_ = cancel.cancelled() => return Err(HealthError::Cancelled),
|
||||
result = disk.disk_info(&options) => result,
|
||||
};
|
||||
let Ok(info) = info else {
|
||||
capacity_complete = false;
|
||||
flags_complete = false;
|
||||
continue;
|
||||
};
|
||||
if !info.fresh_capacity || info.used > info.total {
|
||||
capacity_complete = false;
|
||||
flags_complete = false;
|
||||
continue;
|
||||
}
|
||||
match disk.runtime_state().as_str() {
|
||||
"online" | "returning" => {}
|
||||
"offline" => {
|
||||
flags.insert("cluster.degraded".to_owned());
|
||||
flags.insert("drive.offline".to_owned());
|
||||
}
|
||||
_ => {
|
||||
flags.insert("cluster.degraded".to_owned());
|
||||
}
|
||||
}
|
||||
if include_capacity {
|
||||
match (
|
||||
total_bytes.checked_add(info.total).filter(|value| *value <= MAX_SAFE_INTEGER),
|
||||
used_bytes.checked_add(info.used).filter(|value| *value <= MAX_SAFE_INTEGER),
|
||||
) {
|
||||
(Some(total), Some(used)) => {
|
||||
total_bytes = total;
|
||||
used_bytes = used;
|
||||
}
|
||||
_ => capacity_complete = false,
|
||||
}
|
||||
}
|
||||
if info.healing {
|
||||
flags.insert("cluster.healing".to_owned());
|
||||
}
|
||||
}
|
||||
|
||||
Ok(HealthSourceObservation {
|
||||
observed_at_unix: Some(observed_at_unix),
|
||||
capacity_complete,
|
||||
total_bytes: capacity_complete.then_some(total_bytes),
|
||||
used_bytes: capacity_complete.then_some(used_bytes),
|
||||
coarse_flags_complete: flags_complete,
|
||||
coarse_flags: flags_complete.then(|| flags.into_iter().collect()),
|
||||
})
|
||||
}
|
||||
|
||||
pub fn sign_health_export(
|
||||
request: &HealthServiceRequest,
|
||||
result: &HealthDiagnosticResult,
|
||||
identity: &DeviceIdentity,
|
||||
cancel: &CancellationToken,
|
||||
) -> Result<SignedHealthExport, HealthError> {
|
||||
validate_request(request, Utc::now().timestamp())?;
|
||||
check_cancel(cancel)?;
|
||||
if result.schema_version != HEALTH_SCHEMA_VERSION
|
||||
|| result.run_uid != request.run_uid
|
||||
|| result.tool_id != HEALTH_TOOL_ID
|
||||
|| result.capability != HEALTH_SERVICE_CAPABILITY
|
||||
{
|
||||
return Err(HealthError::InvalidRequest);
|
||||
}
|
||||
let result_json = serde_json::to_vec(result).map_err(|_| HealthError::Encoding)?;
|
||||
if result_json.is_empty() || u64::try_from(result_json.len()).unwrap_or(u64::MAX) > MAX_HEALTH_OUTPUT_BYTES {
|
||||
return Err(HealthError::LimitExceeded);
|
||||
}
|
||||
let device_key_id = hex_lower(&Sha256::digest(identity.public_key_der()));
|
||||
let result_sha256 = hex_lower(&Sha256::digest(&result_json));
|
||||
let envelope = HealthEnvelope {
|
||||
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: HEALTH_TOOL_ID,
|
||||
schema_version: HEALTH_SCHEMA_VERSION,
|
||||
classification: "L0",
|
||||
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: HealthPayload {
|
||||
path: RESULT_PATH,
|
||||
media_type: "application/json",
|
||||
size_bytes: u64::try_from(result_json.len()).unwrap_or(u64::MAX),
|
||||
sha256: &result_sha256,
|
||||
},
|
||||
};
|
||||
let envelope_json = serde_json::to_vec(&envelope).map_err(|_| HealthError::Encoding)?;
|
||||
if envelope_json.is_empty() || envelope_json.len() > MAX_ENVELOPE_BYTES {
|
||||
return Err(HealthError::LimitExceeded);
|
||||
}
|
||||
let envelope_signature = signature_document(identity, &device_key_id, &envelope_json)?;
|
||||
check_cancel(cancel)?;
|
||||
let archive_bytes = archive(&envelope_json, &envelope_signature, &result_json)?;
|
||||
if u64::try_from(archive_bytes.len()).unwrap_or(u64::MAX) > MAX_HEALTH_OUTPUT_BYTES {
|
||||
return Err(HealthError::LimitExceeded);
|
||||
}
|
||||
Ok(SignedHealthExport {
|
||||
artifact_uid: request.artifact_uid.clone(),
|
||||
outcome: result.outcome,
|
||||
reason_code: result.reason_code,
|
||||
envelope_json,
|
||||
envelope_signature,
|
||||
result_json,
|
||||
archive_sha256: hex_lower(&Sha256::digest(&archive_bytes)),
|
||||
archive_bytes,
|
||||
})
|
||||
}
|
||||
|
||||
fn result_reason(freshness: HealthFreshness, observation: &HealthSourceObservation) -> HealthResultReason {
|
||||
match freshness {
|
||||
HealthFreshness::Stale => HealthResultReason::EvidenceStale,
|
||||
HealthFreshness::Unknown if observation.observed_at_unix.is_some() => HealthResultReason::ClockSkew,
|
||||
HealthFreshness::Unknown => HealthResultReason::EvidenceFreshnessUnknown,
|
||||
HealthFreshness::Current if !observation.capacity_complete || !observation.coarse_flags_complete => {
|
||||
HealthResultReason::EvidenceIncomplete
|
||||
}
|
||||
HealthFreshness::Current => HealthResultReason::CatalogPartial,
|
||||
}
|
||||
}
|
||||
|
||||
fn capacity_check(observation: &HealthSourceObservation, freshness: HealthFreshness) -> HealthCheckResult {
|
||||
let (outcome, reason_code) = match freshness {
|
||||
HealthFreshness::Stale => (HealthRuleOutcome::Unknown, "EVIDENCE_STALE"),
|
||||
HealthFreshness::Unknown if observation.observed_at_unix.is_some() => (HealthRuleOutcome::Unknown, "CLOCK_SKEW"),
|
||||
HealthFreshness::Unknown => (HealthRuleOutcome::Unknown, "EVIDENCE_FRESHNESS_UNKNOWN"),
|
||||
HealthFreshness::Current if !observation.capacity_complete => (HealthRuleOutcome::Unknown, "EVIDENCE_MISSING"),
|
||||
HealthFreshness::Current => match (observation.total_bytes, observation.used_bytes) {
|
||||
(Some(0), Some(_)) => (HealthRuleOutcome::Unknown, "CAPACITY_UNAVAILABLE"),
|
||||
(Some(total), Some(used)) if used > total => (HealthRuleOutcome::Unknown, "INVALID_EVIDENCE"),
|
||||
(Some(total), Some(used)) if used == total => (HealthRuleOutcome::Fail, "CAPACITY_EXHAUSTED"),
|
||||
(Some(total), Some(used)) if used < total => (HealthRuleOutcome::Pass, "CAPACITY_REMAINING"),
|
||||
_ => (HealthRuleOutcome::Unknown, "EVIDENCE_MISSING"),
|
||||
},
|
||||
};
|
||||
HealthCheckResult {
|
||||
check_id: "capacity.exhausted",
|
||||
outcome,
|
||||
reason_code,
|
||||
resource_alias: RESOURCE_ALIAS,
|
||||
}
|
||||
}
|
||||
|
||||
fn flags_check(observation: &HealthSourceObservation, freshness: HealthFreshness) -> HealthCheckResult {
|
||||
let (outcome, reason_code) = match freshness {
|
||||
HealthFreshness::Stale => (HealthRuleOutcome::Unknown, "EVIDENCE_STALE"),
|
||||
HealthFreshness::Unknown if observation.observed_at_unix.is_some() => (HealthRuleOutcome::Unknown, "CLOCK_SKEW"),
|
||||
HealthFreshness::Unknown => (HealthRuleOutcome::Unknown, "EVIDENCE_FRESHNESS_UNKNOWN"),
|
||||
HealthFreshness::Current if !observation.coarse_flags_complete => (HealthRuleOutcome::Unknown, "EVIDENCE_MISSING"),
|
||||
HealthFreshness::Current => match observation.coarse_flags.as_deref() {
|
||||
Some(flags) if flags.iter().any(|flag| !ALLOWED_FLAGS.contains(&flag.as_str())) => {
|
||||
(HealthRuleOutcome::Unknown, "INVALID_EVIDENCE")
|
||||
}
|
||||
Some(flags) if flags.is_empty() => (HealthRuleOutcome::Pass, "NO_COARSE_CONDITION_REPORTED"),
|
||||
Some(_) => (HealthRuleOutcome::Fail, "COARSE_CONDITION_REPORTED"),
|
||||
None => (HealthRuleOutcome::Unknown, "EVIDENCE_MISSING"),
|
||||
},
|
||||
};
|
||||
HealthCheckResult {
|
||||
check_id: "cluster.reportedFlags",
|
||||
outcome,
|
||||
reason_code,
|
||||
resource_alias: RESOURCE_ALIAS,
|
||||
}
|
||||
}
|
||||
|
||||
fn freshness(observed_at: Option<DateTime<Utc>>, evaluated_at: DateTime<Utc>, max_age: u64) -> HealthFreshness {
|
||||
let Some(observed_at) = observed_at else {
|
||||
return HealthFreshness::Unknown;
|
||||
};
|
||||
let Ok(age) = evaluated_at.signed_duration_since(observed_at).to_std() else {
|
||||
return HealthFreshness::Unknown;
|
||||
};
|
||||
if age.as_secs() > max_age {
|
||||
HealthFreshness::Stale
|
||||
} else {
|
||||
HealthFreshness::Current
|
||||
}
|
||||
}
|
||||
|
||||
fn validate_request(request: &HealthServiceRequest, now: i64) -> Result<(), HealthError> {
|
||||
if request.schema_version != HEALTH_SCHEMA_VERSION || request.capability != HEALTH_SERVICE_CAPABILITY {
|
||||
return Err(HealthError::Unsupported);
|
||||
}
|
||||
if !request.consent.active || request.consent.policy_revision == 0 {
|
||||
return Err(HealthError::ConsentRequired);
|
||||
}
|
||||
if request.consent.expires_at_unix <= now || request.expires_at_unix > request.consent.expires_at_unix {
|
||||
return Err(HealthError::ConsentExpired);
|
||||
}
|
||||
if request.produced_at_unix > now.saturating_add(300)
|
||||
|| request.expires_at_unix <= now
|
||||
|| request.expires_at_unix <= request.produced_at_unix
|
||||
{
|
||||
return Err(HealthError::Expired);
|
||||
}
|
||||
if !(1..=MAX_EVIDENCE_AGE_SECONDS).contains(&request.max_evidence_age_seconds)
|
||||
|| !uuid7(&request.run_uid)
|
||||
|| !uuid7(&request.artifact_uid)
|
||||
|| !uuid7(&request.consent.consent_uid)
|
||||
|| !resource_names_match(request)
|
||||
|| !request.provenance.is_valid()
|
||||
{
|
||||
return Err(HealthError::InvalidRequest);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn check_cancel(cancel: &CancellationToken) -> Result<(), HealthError> {
|
||||
if cancel.is_cancelled() {
|
||||
Err(HealthError::Cancelled)
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct HealthEnvelope<'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: HealthPayload<'a>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct HealthPayload<'a> {
|
||||
path: &'static str,
|
||||
media_type: &'static str,
|
||||
size_bytes: u64,
|
||||
sha256: &'a str,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct HealthSignature<'a> {
|
||||
algorithm: &'static str,
|
||||
key_id: &'a str,
|
||||
value: String,
|
||||
}
|
||||
|
||||
fn signature_document(identity: &DeviceIdentity, key_id: &str, envelope: &[u8]) -> Result<Vec<u8>, HealthError> {
|
||||
let pkcs8 = identity.to_pkcs8_der().map_err(|_| HealthError::Signing)?;
|
||||
let signing_key = SigningKey::from_pkcs8_der(pkcs8.as_slice()).map_err(|_| HealthError::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(&HealthSignature {
|
||||
algorithm: "ES256",
|
||||
key_id,
|
||||
value: URL_SAFE_NO_PAD.encode_to_string(signature.normalize_s().to_bytes()),
|
||||
})
|
||||
.map_err(|_| HealthError::Encoding)
|
||||
}
|
||||
|
||||
fn archive(envelope: &[u8], signature: &[u8], result: &[u8]) -> Result<Vec<u8>, HealthError> {
|
||||
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(0o600);
|
||||
for (name, bytes) in [(ENVELOPE_PATH, envelope), (SIGNATURE_PATH, signature), (RESULT_PATH, result)] {
|
||||
writer.start_file(name, options).map_err(|_| HealthError::Encoding)?;
|
||||
writer.write_all(bytes).map_err(|_| HealthError::Encoding)?;
|
||||
}
|
||||
writer
|
||||
.finish()
|
||||
.map(|cursor| cursor.into_inner())
|
||||
.map_err(|_| HealthError::Encoding)
|
||||
}
|
||||
|
||||
fn timestamp(value: i64) -> Result<String, HealthError> {
|
||||
DateTime::<Utc>::from_timestamp(value, 0)
|
||||
.map(|value| value.to_rfc3339_opts(SecondsFormat::Secs, true))
|
||||
.ok_or(HealthError::InvalidRequest)
|
||||
}
|
||||
|
||||
fn resource_names_match(request: &HealthServiceRequest) -> 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 hex_lower(bytes: &[u8]) -> String {
|
||||
bytes.iter().fold(String::with_capacity(bytes.len() * 2), |mut value, byte| {
|
||||
use std::fmt::Write as _;
|
||||
let _ = write!(value, "{byte:02x}");
|
||||
value
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn collector_lease_limits_health_jobs_to_one() {
|
||||
let active = CollectorLease::acquire().expect("first health collector");
|
||||
assert!(matches!(CollectorLease::acquire(), Err(HealthError::Busy)));
|
||||
drop(active);
|
||||
CollectorLease::acquire().expect("collector lease should be released");
|
||||
}
|
||||
}
|
||||
@@ -41,6 +41,11 @@ use super::{
|
||||
capture_top_rpc, encode_signed_profile_export, measure_drive, measure_network, runtime_network_peer_aliases,
|
||||
sign_drive_export, sign_network_export, sign_top_export_with_nonce,
|
||||
};
|
||||
use super::{
|
||||
HEALTH_SCHEMA_VERSION, HEALTH_SERVICE_CAPABILITY, HEALTH_TIMEOUT_SECONDS, HealthError, HealthServiceRequest,
|
||||
LocalHealthConsent, MAX_EVIDENCE_AGE_SECONDS, MAX_HEALTH_CPU_MILLIS, MAX_HEALTH_MEMORY_BYTES, MAX_HEALTH_OUTPUT_BYTES,
|
||||
collect_runtime_health,
|
||||
};
|
||||
use crate::connect::DeviceIdentity;
|
||||
|
||||
const PROTOCOL_VERSION: &str = "v1";
|
||||
@@ -52,6 +57,7 @@ const PERFORMANCE_NETWORK_JOB_TYPE: &str = "performance.network";
|
||||
const TOP_API_JOB_TYPE: &str = "top.api";
|
||||
const TOP_LOCKS_JOB_TYPE: &str = "top.locks";
|
||||
const TOP_RPC_JOB_TYPE: &str = "top.rpc";
|
||||
const HEALTH_JOB_TYPE: &str = "health.check";
|
||||
pub const DIAGNOSTIC_JOB_SIGNATURE_DOMAIN: &[u8] = b"rustfs-connect-agent-job-v1\0";
|
||||
const MAX_JOB_LIFETIME_SECONDS: i64 = 1_800;
|
||||
const MAX_FUTURE_SKEW_SECONDS: i64 = 300;
|
||||
@@ -76,6 +82,7 @@ const MIN_TOP_RPC_MEMORY_BYTES: u64 = 1_048_576;
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
enum DiagnosticJobKind {
|
||||
Health,
|
||||
ProfileCpu,
|
||||
ProfileThreads,
|
||||
ProfileMemory,
|
||||
@@ -109,9 +116,13 @@ pub struct DiagnosticJobParameters {
|
||||
pub consent_uid: String,
|
||||
pub consent_policy_revision: u64,
|
||||
pub consent_expires_at: String,
|
||||
#[serde(default, skip_serializing_if = "is_zero")]
|
||||
pub duration_millis: u64,
|
||||
#[serde(default, skip_serializing_if = "is_zero")]
|
||||
pub sample_period_micros: u64,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub max_evidence_age_seconds: Option<u64>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub traffic_bytes: Option<u64>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub target_alias: Option<String>,
|
||||
@@ -415,13 +426,29 @@ impl DiagnosticJobEnvelope {
|
||||
|| self.limits.max_memory_bytes > MAX_MEMORY_BYTES
|
||||
|| self.limits.max_cpu_millis == 0
|
||||
|| self.limits.max_cpu_millis > MAX_CPU_MILLIS
|
||||
|| self.parameters.duration_millis == 0
|
||||
{
|
||||
return Err(DiagnosticJobError::LimitExceeded);
|
||||
}
|
||||
if kind == DiagnosticJobKind::Health {
|
||||
if self.limits.timeout_seconds != HEALTH_TIMEOUT_SECONDS
|
||||
|| self.limits.max_output_bytes != MAX_HEALTH_OUTPUT_BYTES
|
||||
|| self.limits.max_memory_bytes != MAX_HEALTH_MEMORY_BYTES
|
||||
|| self.limits.max_cpu_millis != MAX_HEALTH_CPU_MILLIS
|
||||
|| self.parameters.duration_millis != 0
|
||||
|| self.parameters.sample_period_micros != 0
|
||||
|| !matches!(self.parameters.max_evidence_age_seconds, Some(1..=MAX_EVIDENCE_AGE_SECONDS))
|
||||
{
|
||||
return Err(DiagnosticJobError::LimitExceeded);
|
||||
}
|
||||
} else if self.parameters.duration_millis == 0
|
||||
|| self.parameters.duration_millis > self.limits.timeout_seconds.saturating_mul(1_000)
|
||||
|| self.parameters.duration_millis > self.limits.max_cpu_millis
|
||||
|| self.parameters.sample_period_micros == 0
|
||||
|| self.parameters.sample_period_micros > self.parameters.duration_millis.saturating_mul(1_000)
|
||||
{
|
||||
return Err(DiagnosticJobError::LimitExceeded);
|
||||
} else if self.parameters.max_evidence_age_seconds.is_some() {
|
||||
return Err(DiagnosticJobError::Invalid);
|
||||
}
|
||||
if kind == DiagnosticJobKind::TopApi
|
||||
&& (self.limits.max_cpu_millis > MAX_TOP_API_CPU_MILLIS || self.limits.max_memory_bytes < MIN_TOP_API_MEMORY_BYTES)
|
||||
@@ -481,6 +508,9 @@ impl DiagnosticJobEnvelope {
|
||||
|
||||
fn kind(&self) -> Result<DiagnosticJobKind, DiagnosticJobError> {
|
||||
match (self.job_type.as_str(), self.required_capabilities.as_slice(), self.schema_version) {
|
||||
(HEALTH_JOB_TYPE, [capability], HEALTH_SCHEMA_VERSION) if capability == HEALTH_SERVICE_CAPABILITY => {
|
||||
Ok(DiagnosticJobKind::Health)
|
||||
}
|
||||
(PROFILE_CPU_JOB_TYPE, [capability], PROFILE_SCHEMA_VERSION) if capability == CPU_PROFILE_CAPABILITY => {
|
||||
Ok(DiagnosticJobKind::ProfileCpu)
|
||||
}
|
||||
@@ -530,6 +560,7 @@ pub async fn execute_diagnostic_job(
|
||||
let nonce = job.nonce;
|
||||
let envelope = job.envelope;
|
||||
match envelope.kind()? {
|
||||
DiagnosticJobKind::Health => execute_health_job(envelope, nonce, identity, provenance, cancel).await,
|
||||
DiagnosticJobKind::ProfileCpu => execute_profile_cpu_job(envelope, nonce, identity, provenance, cancel).await,
|
||||
DiagnosticJobKind::ProfileThreads => execute_profile_threads_job(envelope, nonce, identity, provenance, cancel).await,
|
||||
DiagnosticJobKind::ProfileMemory => execute_profile_memory_job(envelope, nonce, identity, provenance, cancel).await,
|
||||
@@ -551,6 +582,88 @@ pub async fn execute_diagnostic_job(
|
||||
}
|
||||
}
|
||||
|
||||
async fn execute_health_job(
|
||||
envelope: DiagnosticJobEnvelope,
|
||||
nonce: [u8; 32],
|
||||
identity: &DeviceIdentity,
|
||||
provenance: ProfileProvenance,
|
||||
cancel: &CancellationToken,
|
||||
) -> Result<DiagnosticJobExecution, DiagnosticJobError> {
|
||||
let expire = parse_time(&envelope.expire_time)?;
|
||||
let consent_expire = parse_time(&envelope.parameters.consent_expires_at)?;
|
||||
let request = HealthServiceRequest {
|
||||
organization_name: envelope.organization_name,
|
||||
cluster_name: envelope.cluster_name,
|
||||
device_name: envelope.device_name,
|
||||
run_uid: envelope.job_id.clone(),
|
||||
artifact_uid: envelope.parameters.artifact_uid,
|
||||
schema_version: envelope.schema_version,
|
||||
capability: HEALTH_SERVICE_CAPABILITY.to_owned(),
|
||||
consent: LocalHealthConsent {
|
||||
consent_uid: envelope.parameters.consent_uid,
|
||||
policy_revision: envelope.parameters.consent_policy_revision,
|
||||
expires_at_unix: consent_expire.timestamp(),
|
||||
active: true,
|
||||
},
|
||||
produced_at_unix: Utc::now().timestamp(),
|
||||
expires_at_unix: expire.timestamp(),
|
||||
nonce,
|
||||
max_evidence_age_seconds: envelope
|
||||
.parameters
|
||||
.max_evidence_age_seconds
|
||||
.ok_or(DiagnosticJobError::LimitExceeded)?,
|
||||
provenance,
|
||||
};
|
||||
let timeout = Duration::from_secs(envelope.limits.timeout_seconds);
|
||||
let health_cancel = cancel.child_token();
|
||||
let collected = tokio::time::timeout(timeout, collect_runtime_health(&request, identity, &health_cancel)).await;
|
||||
let export = match collected {
|
||||
Ok(Ok(export)) => export,
|
||||
Ok(Err(HealthError::SourceUnavailable)) => {
|
||||
return Ok(DiagnosticJobExecution {
|
||||
job_id: envelope.job_id,
|
||||
outcome: "FAILED".to_owned(),
|
||||
reason: "SOURCE_UNAVAILABLE".to_owned(),
|
||||
artifact_uid: None,
|
||||
artifact_sha256: None,
|
||||
artifact_bytes: None,
|
||||
});
|
||||
}
|
||||
Ok(Err(error)) => return Err(health_failure(error)),
|
||||
Err(_) => {
|
||||
health_cancel.cancel();
|
||||
return Err(DiagnosticJobError::LimitExceeded);
|
||||
}
|
||||
};
|
||||
if export.archive_bytes.len() > usize::try_from(envelope.limits.max_output_bytes).unwrap_or(usize::MAX) {
|
||||
return Err(DiagnosticJobError::LimitExceeded);
|
||||
}
|
||||
Ok(DiagnosticJobExecution {
|
||||
job_id: envelope.job_id,
|
||||
outcome: export.outcome.as_str().to_owned(),
|
||||
reason: export.reason_code.as_str().to_owned(),
|
||||
artifact_uid: Some(export.artifact_uid),
|
||||
artifact_sha256: Some(export.archive_sha256),
|
||||
artifact_bytes: Some(export.archive_bytes),
|
||||
})
|
||||
}
|
||||
|
||||
fn health_failure(error: HealthError) -> DiagnosticJobError {
|
||||
match error {
|
||||
HealthError::Cancelled => DiagnosticJobError::Cancelled,
|
||||
HealthError::LimitExceeded | HealthError::Busy => DiagnosticJobError::LimitExceeded,
|
||||
HealthError::SourceUnavailable | HealthError::CollectionFailed => DiagnosticJobError::CollectionFailed,
|
||||
HealthError::Signing | HealthError::Encoding => DiagnosticJobError::ExportFailed,
|
||||
HealthError::ConsentExpired | HealthError::Expired => DiagnosticJobError::Expired,
|
||||
HealthError::Unsupported => DiagnosticJobError::Unsupported,
|
||||
HealthError::ConsentRequired | HealthError::InvalidRequest => DiagnosticJobError::Invalid,
|
||||
}
|
||||
}
|
||||
|
||||
fn is_zero(value: &u64) -> bool {
|
||||
*value == 0
|
||||
}
|
||||
|
||||
fn runtime_drive_scratch_root() -> Option<std::path::PathBuf> {
|
||||
let endpoint_pools = crate::runtime_sources::current_endpoints_handle()?;
|
||||
drive_scratch_root_from_endpoints(&endpoint_pools)
|
||||
@@ -1233,6 +1346,7 @@ mod tests {
|
||||
consent_expires_at: "2030-01-01T00:01:00Z".to_owned(),
|
||||
duration_millis: 1_000,
|
||||
sample_period_micros: 10_000,
|
||||
max_evidence_age_seconds: None,
|
||||
traffic_bytes: None,
|
||||
target_alias: None,
|
||||
scratch_bytes: None,
|
||||
@@ -1275,6 +1389,7 @@ mod tests {
|
||||
use clap::CommandFactory;
|
||||
|
||||
let expected = [
|
||||
("health.check.service@1", &[][..]),
|
||||
("performance.client@1", &["performance", "client"][..]),
|
||||
("performance.drive@1", &["performance", "drive"][..]),
|
||||
("performance.network@1", &[][..]),
|
||||
@@ -1306,6 +1421,7 @@ mod tests {
|
||||
if path.is_empty() {
|
||||
let mut job = envelope();
|
||||
let (job_type, kind) = match capability {
|
||||
HEALTH_SERVICE_CAPABILITY => (HEALTH_JOB_TYPE, DiagnosticJobKind::Health),
|
||||
NETWORK_CAPABILITY => (PERFORMANCE_NETWORK_JOB_TYPE, DiagnosticJobKind::PerformanceNetwork),
|
||||
MEMORY_PROFILE_SERVICE_CAPABILITY => (PROFILE_MEMORY_JOB_TYPE, DiagnosticJobKind::ProfileMemory),
|
||||
_ => panic!("{capability} is missing service dispatch"),
|
||||
@@ -1313,6 +1429,17 @@ mod tests {
|
||||
job.job_type = job_type.to_owned();
|
||||
job.required_capabilities = vec![capability.to_owned()];
|
||||
job.schema_version = NETWORK_SCHEMA_VERSION;
|
||||
if capability == HEALTH_SERVICE_CAPABILITY {
|
||||
job.limits = DiagnosticJobLimits {
|
||||
timeout_seconds: HEALTH_TIMEOUT_SECONDS,
|
||||
max_output_bytes: MAX_HEALTH_OUTPUT_BYTES,
|
||||
max_memory_bytes: MAX_HEALTH_MEMORY_BYTES,
|
||||
max_cpu_millis: MAX_HEALTH_CPU_MILLIS,
|
||||
};
|
||||
job.parameters.duration_millis = 0;
|
||||
job.parameters.sample_period_micros = 0;
|
||||
job.parameters.max_evidence_age_seconds = Some(300);
|
||||
}
|
||||
assert_eq!(job.kind(), Ok(kind));
|
||||
continue;
|
||||
}
|
||||
@@ -1333,6 +1460,70 @@ mod tests {
|
||||
.expect("valid job");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn accepts_only_the_bounded_health_service_contract() {
|
||||
let mut health = envelope();
|
||||
health.job_type = HEALTH_JOB_TYPE.to_owned();
|
||||
health.required_capabilities = vec![HEALTH_SERVICE_CAPABILITY.to_owned()];
|
||||
health.limits = DiagnosticJobLimits {
|
||||
timeout_seconds: HEALTH_TIMEOUT_SECONDS,
|
||||
max_output_bytes: MAX_HEALTH_OUTPUT_BYTES,
|
||||
max_memory_bytes: MAX_HEALTH_MEMORY_BYTES,
|
||||
max_cpu_millis: MAX_HEALTH_CPU_MILLIS,
|
||||
};
|
||||
health.parameters.duration_millis = 0;
|
||||
health.parameters.sample_period_micros = 0;
|
||||
health.parameters.max_evidence_age_seconds = Some(300);
|
||||
let (health, signer) = signed_envelope(health);
|
||||
signer
|
||||
.verify(&health, &target(&health), "2030-01-01T00:00:10Z".parse().expect("time"))
|
||||
.expect("valid health service job");
|
||||
|
||||
let unsigned = serde_json::to_value(health.unsigned()).expect("unsigned health job");
|
||||
assert!(unsigned["parameters"].get("durationMillis").is_none());
|
||||
assert!(unsigned["parameters"].get("samplePeriodMicros").is_none());
|
||||
assert_eq!(unsigned["parameters"]["maxEvidenceAgeSeconds"], 300);
|
||||
|
||||
let mut old_capability = health.clone();
|
||||
old_capability.required_capabilities = vec!["health.check@1".to_owned()];
|
||||
assert_eq!(
|
||||
signer.verify(&old_capability, &target(&old_capability), "2030-01-01T00:00:10Z".parse().expect("time")),
|
||||
Err(DiagnosticJobError::Unsupported)
|
||||
);
|
||||
|
||||
let mut oversized = health.clone();
|
||||
oversized.limits.max_output_bytes += 1;
|
||||
let (oversized, oversized_signer) = signed_envelope(oversized);
|
||||
assert_eq!(
|
||||
oversized_signer.verify(&oversized, &target(&oversized), "2030-01-01T00:00:10Z".parse().expect("time")),
|
||||
Err(DiagnosticJobError::LimitExceeded)
|
||||
);
|
||||
|
||||
let mut missing_freshness = health.clone();
|
||||
missing_freshness.parameters.max_evidence_age_seconds = None;
|
||||
let (missing_freshness, missing_signer) = signed_envelope(missing_freshness);
|
||||
assert_eq!(
|
||||
missing_signer.verify(
|
||||
&missing_freshness,
|
||||
&target(&missing_freshness),
|
||||
"2030-01-01T00:00:10Z".parse().expect("time")
|
||||
),
|
||||
Err(DiagnosticJobError::LimitExceeded)
|
||||
);
|
||||
|
||||
let mut profile_parameter = health;
|
||||
profile_parameter.parameters.duration_millis = 1;
|
||||
let (profile_parameter, profile_parameter_signer) = signed_envelope(profile_parameter);
|
||||
assert_eq!(
|
||||
profile_parameter_signer.verify(
|
||||
&profile_parameter,
|
||||
&target(&profile_parameter),
|
||||
"2030-01-01T00:00:10Z".parse().expect("time")
|
||||
),
|
||||
Err(DiagnosticJobError::LimitExceeded)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn accepts_only_the_thread_profile_capability_pair() {
|
||||
let mut threads = envelope();
|
||||
|
||||
@@ -12,6 +12,7 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
mod health;
|
||||
mod inspect;
|
||||
mod job;
|
||||
pub(crate) mod job_delivery;
|
||||
@@ -40,6 +41,7 @@ mod trace_runtime;
|
||||
|
||||
/// Signed diagnostic producers available through the CLI or authenticated service jobs.
|
||||
pub const CONNECT_DIAGNOSTIC_CAPABILITIES: &[&str] = &[
|
||||
health::HEALTH_SERVICE_CAPABILITY,
|
||||
perf_client::CLIENT_CAPABILITY,
|
||||
perf_drive::DRIVE_CAPABILITY,
|
||||
perf_network::NETWORK_CAPABILITY,
|
||||
@@ -64,6 +66,13 @@ pub const CONNECT_DIAGNOSTIC_CAPABILITIES: &[&str] = &[
|
||||
#[path = "trace_runtime_unsupported.rs"]
|
||||
mod trace_runtime;
|
||||
|
||||
pub use health::{
|
||||
HEALTH_CATALOG_CHECKS, HEALTH_SCHEMA_VERSION, HEALTH_SERVICE_CAPABILITY, HEALTH_TIMEOUT_SECONDS, HealthCheckResult,
|
||||
HealthDiagnosticResult, HealthError, HealthFreshness, HealthOutcome, HealthResultReason, HealthRuleOutcome,
|
||||
HealthServiceRequest, HealthSourceObservation, LocalHealthConsent, MAX_EVIDENCE_AGE_SECONDS, MAX_HEALTH_CPU_MILLIS,
|
||||
MAX_HEALTH_MEMORY_BYTES, MAX_HEALTH_OUTPUT_BYTES, SignedHealthExport, collect_runtime_health, evaluate_health_observation,
|
||||
sign_health_export,
|
||||
};
|
||||
pub use inspect::{
|
||||
INSPECT_CAPABILITY, INSPECT_SCHEMA_VERSION, InspectArtifactConsent, InspectDiagnosticResult, InspectError, InspectFinding,
|
||||
InspectOutcome, InspectProvenance, InspectReason, InspectReasonCode, InspectRequest, InspectRule, InspectRuleOutcome,
|
||||
|
||||
@@ -185,6 +185,14 @@ impl ProfileProvenance {
|
||||
pub(crate) fn build_features(&self) -> &[String] {
|
||||
&self.build_features
|
||||
}
|
||||
|
||||
pub(crate) fn is_valid(&self) -> bool {
|
||||
lower_hex(&self.source_commit, 40)
|
||||
&& lower_hex(&self.executable_sha256, 64)
|
||||
&& version(&self.rustfs_version)
|
||||
&& self.build_features.len() <= MAX_BUILD_FEATURES
|
||||
&& self.build_features.iter().all(|feature| build_feature(feature))
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
|
||||
@@ -969,11 +977,7 @@ impl ProfileCaptureRequest {
|
||||
|| !uuid7(&self.artifact_uid)
|
||||
|| !uuid7(&self.consent.consent_uid)
|
||||
|| !resource_names_match(self)
|
||||
|| !lower_hex(&self.provenance.source_commit, 40)
|
||||
|| !lower_hex(&self.provenance.executable_sha256, 64)
|
||||
|| !version(&self.provenance.rustfs_version)
|
||||
|| self.provenance.build_features.len() > MAX_BUILD_FEATURES
|
||||
|| !self.provenance.build_features.iter().all(|feature| build_feature(feature))
|
||||
|| !self.provenance.is_valid()
|
||||
{
|
||||
return Err(ProfileError::InvalidRequest);
|
||||
}
|
||||
|
||||
@@ -86,6 +86,13 @@ pub use diagnostics::{
|
||||
DIAGNOSTIC_JOB_SIGNATURE_DOMAIN, DiagnosticJobEnvelope, DiagnosticJobError, DiagnosticJobExecution, DiagnosticJobLimits,
|
||||
DiagnosticJobParameters, DiagnosticJobTarget, TrustedDiagnosticJobSigner, VerifiedDiagnosticJob, execute_diagnostic_job,
|
||||
};
|
||||
pub use diagnostics::{
|
||||
HEALTH_CATALOG_CHECKS, HEALTH_SCHEMA_VERSION, HEALTH_SERVICE_CAPABILITY, HEALTH_TIMEOUT_SECONDS, HealthCheckResult,
|
||||
HealthDiagnosticResult, HealthError, HealthFreshness, HealthOutcome, HealthResultReason, HealthRuleOutcome,
|
||||
HealthServiceRequest, HealthSourceObservation, LocalHealthConsent, MAX_EVIDENCE_AGE_SECONDS, MAX_HEALTH_CPU_MILLIS,
|
||||
MAX_HEALTH_MEMORY_BYTES, MAX_HEALTH_OUTPUT_BYTES, SignedHealthExport, collect_runtime_health, evaluate_health_observation,
|
||||
sign_health_export,
|
||||
};
|
||||
pub use diagnostics::{
|
||||
LocalNetworkConsent, MAX_NETWORK_ARCHIVE_BYTES, MAX_NETWORK_BANDWIDTH_BYTES_PER_SECOND, MAX_NETWORK_DECOMPRESSED_BYTES,
|
||||
MAX_NETWORK_DURATION, MAX_NETWORK_ENVELOPE_BYTES, MAX_NETWORK_OPERATIONS, MAX_NETWORK_PEERS, MAX_NETWORK_RESULT_BYTES,
|
||||
|
||||
@@ -59,7 +59,7 @@ fn bucket_targets_metadata_lock_shard(bucket: &str) -> usize {
|
||||
|
||||
pub(crate) mod contract {
|
||||
pub(crate) mod admin {
|
||||
pub(crate) use super::super::storage_contracts::StorageAdminApi;
|
||||
pub(crate) use super::super::storage_contracts::{DiskSetSelector, StorageAdminApi};
|
||||
}
|
||||
|
||||
pub(crate) mod bucket {
|
||||
|
||||
@@ -0,0 +1,274 @@
|
||||
// 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::io::Read as _;
|
||||
use std::time::Duration;
|
||||
|
||||
use base64_simd::URL_SAFE_NO_PAD;
|
||||
use chrono::{TimeDelta, Utc};
|
||||
use p256::ecdsa::signature::Verifier as _;
|
||||
use p256::ecdsa::{Signature, VerifyingKey};
|
||||
use p256::pkcs8::DecodePublicKey as _;
|
||||
use rustfs::connect::{
|
||||
DeviceIdentity, HEALTH_CATALOG_CHECKS, HEALTH_SCHEMA_VERSION, HEALTH_SERVICE_CAPABILITY, HealthError, HealthServiceRequest,
|
||||
HealthSourceObservation, LocalHealthConsent, ProfileProvenance, collect_runtime_health, evaluate_health_observation,
|
||||
sign_health_export,
|
||||
};
|
||||
use serde_json::Value;
|
||||
use sha2::{Digest as _, Sha256};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
const SIGNATURE_DOMAIN: &[u8] = b"rustfs-diagnostic-envelope-v1\0";
|
||||
|
||||
fn request(now: i64) -> HealthServiceRequest {
|
||||
HealthServiceRequest {
|
||||
organization_name: "organizations/018cc251-f400-7abc-8def-0123456789ab".to_owned(),
|
||||
cluster_name: "organizations/018cc251-f400-7abc-8def-0123456789ab/clusters/018cc251-f400-7abc-8def-0123456789ac"
|
||||
.to_owned(),
|
||||
device_name: "organizations/018cc251-f400-7abc-8def-0123456789ab/clusters/018cc251-f400-7abc-8def-0123456789ac/clusterDevices/018cc251-f400-7abc-8def-0123456789ad".to_owned(),
|
||||
run_uid: "018cc251-f400-7abc-8def-0123456789ab".to_owned(),
|
||||
artifact_uid: "018cc251-f400-7abc-8def-0123456789ae".to_owned(),
|
||||
schema_version: HEALTH_SCHEMA_VERSION,
|
||||
capability: HEALTH_SERVICE_CAPABILITY.to_owned(),
|
||||
consent: LocalHealthConsent {
|
||||
consent_uid: "018cc251-f400-7abc-8def-0123456789af".to_owned(),
|
||||
policy_revision: 1,
|
||||
expires_at_unix: now + 120,
|
||||
active: true,
|
||||
},
|
||||
produced_at_unix: now,
|
||||
expires_at_unix: now + 60,
|
||||
nonce: [7; 32],
|
||||
max_evidence_age_seconds: 300,
|
||||
provenance: ProfileProvenance::new("a".repeat(40), "b".repeat(64), "1.0.1", vec![]),
|
||||
}
|
||||
}
|
||||
|
||||
fn observation(now: i64, total: u64, used: u64, flags: Vec<&str>) -> HealthSourceObservation {
|
||||
HealthSourceObservation {
|
||||
observed_at_unix: Some(now),
|
||||
capacity_complete: true,
|
||||
total_bytes: Some(total),
|
||||
used_bytes: Some(used),
|
||||
coarse_flags_complete: true,
|
||||
coarse_flags: Some(flags.into_iter().map(str::to_owned).collect()),
|
||||
}
|
||||
}
|
||||
|
||||
fn json_result(
|
||||
request: &HealthServiceRequest,
|
||||
observation: HealthSourceObservation,
|
||||
evaluated_at: chrono::DateTime<Utc>,
|
||||
) -> Value {
|
||||
serde_json::to_value(evaluate_health_observation(request, observation, evaluated_at, Duration::from_millis(25)))
|
||||
.expect("health result")
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn health_rules_keep_exact_boundaries_and_full_catalog_coverage() {
|
||||
let now = Utc::now();
|
||||
let request = request(now.timestamp());
|
||||
|
||||
let remaining = json_result(&request, observation(now.timestamp(), 100, 99, vec![]), now);
|
||||
assert_eq!(remaining["data"]["checks"][0]["outcome"], "PASS");
|
||||
assert_eq!(remaining["data"]["checks"][0]["reasonCode"], "CAPACITY_REMAINING");
|
||||
assert_eq!(remaining["data"]["checks"][1]["outcome"], "PASS");
|
||||
assert_eq!(remaining["coverage"]["requestedUnits"], HEALTH_CATALOG_CHECKS);
|
||||
assert_eq!(remaining["coverage"]["completedUnits"], 2);
|
||||
assert_eq!(remaining["coverage"]["unsupportedUnits"], 11);
|
||||
assert_eq!(remaining["data"]["unsupportedChecks"].as_array().expect("unsupported").len(), 11);
|
||||
assert_eq!(remaining["data"]["unsupportedChecks"][0]["checkId"], "memory.reportedPressure");
|
||||
|
||||
let exhausted = json_result(&request, observation(now.timestamp(), 100, 100, vec![]), now);
|
||||
assert_eq!(exhausted["data"]["checks"][0]["outcome"], "FAIL");
|
||||
assert_eq!(exhausted["data"]["checks"][0]["reasonCode"], "CAPACITY_EXHAUSTED");
|
||||
|
||||
let zero = json_result(&request, observation(now.timestamp(), 0, 0, vec![]), now);
|
||||
assert_eq!(zero["data"]["checks"][0]["outcome"], "UNKNOWN");
|
||||
assert_eq!(zero["data"]["checks"][0]["reasonCode"], "CAPACITY_UNAVAILABLE");
|
||||
|
||||
let invalid = json_result(&request, observation(now.timestamp(), 99, 100, vec![]), now);
|
||||
assert_eq!(invalid["data"]["checks"][0]["outcome"], "UNKNOWN");
|
||||
assert_eq!(invalid["data"]["checks"][0]["reasonCode"], "INVALID_EVIDENCE");
|
||||
|
||||
let maximum = json_result(
|
||||
&request,
|
||||
observation(now.timestamp(), 9_007_199_254_740_991, 9_007_199_254_740_991, vec![]),
|
||||
now,
|
||||
);
|
||||
assert_eq!(maximum["data"]["checks"][0]["outcome"], "FAIL");
|
||||
assert_eq!(maximum["data"]["checks"][0]["reasonCode"], "CAPACITY_EXHAUSTED");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cancelled_health_collection_stops_before_runtime_access() {
|
||||
let now = Utc::now();
|
||||
let cancel = CancellationToken::new();
|
||||
cancel.cancel();
|
||||
assert_eq!(
|
||||
collect_runtime_health(&request(now.timestamp()), &DeviceIdentity::generate(), &cancel).await,
|
||||
Err(HealthError::Cancelled)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn health_freshness_requires_real_time_and_has_no_silent_default() {
|
||||
let now = Utc::now();
|
||||
let request = request(now.timestamp());
|
||||
|
||||
let equality = json_result(&request, observation((now - TimeDelta::seconds(300)).timestamp(), 100, 50, vec![]), now);
|
||||
assert_eq!(equality["data"]["observations"]["capacity"]["freshness"], "CURRENT");
|
||||
|
||||
let stale = json_result(&request, observation((now - TimeDelta::seconds(301)).timestamp(), 100, 50, vec![]), now);
|
||||
assert_eq!(stale["reasonCode"], "EVIDENCE_STALE");
|
||||
assert_eq!(stale["data"]["checks"][0]["reasonCode"], "EVIDENCE_STALE");
|
||||
|
||||
let future = json_result(&request, observation((now + TimeDelta::seconds(1)).timestamp(), 100, 50, vec![]), now);
|
||||
assert_eq!(future["reasonCode"], "CLOCK_SKEW");
|
||||
assert_eq!(future["data"]["checks"][0]["reasonCode"], "CLOCK_SKEW");
|
||||
|
||||
let missing = json_result(
|
||||
&request,
|
||||
HealthSourceObservation {
|
||||
observed_at_unix: None,
|
||||
capacity_complete: false,
|
||||
total_bytes: None,
|
||||
used_bytes: None,
|
||||
coarse_flags_complete: false,
|
||||
coarse_flags: None,
|
||||
},
|
||||
now,
|
||||
);
|
||||
assert_eq!(missing["reasonCode"], "EVIDENCE_FRESHNESS_UNKNOWN");
|
||||
assert_eq!(missing["data"]["checks"][1]["outcome"], "UNKNOWN");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn every_allowed_coarse_flag_is_a_fixed_failure_without_raw_source_identity() {
|
||||
let now = Utc::now();
|
||||
let request = request(now.timestamp());
|
||||
for flag in [
|
||||
"capacity.critical",
|
||||
"capacity.warning",
|
||||
"clock.skew",
|
||||
"cluster.degraded",
|
||||
"cluster.healing",
|
||||
"cluster.readonly",
|
||||
"drive.offline",
|
||||
"node.offline",
|
||||
] {
|
||||
let result = json_result(&request, observation(now.timestamp(), 100, 50, vec![flag]), now);
|
||||
assert_eq!(result["data"]["checks"][1]["outcome"], "FAIL", "{flag}");
|
||||
assert_eq!(result["data"]["checks"][1]["reasonCode"], "COARSE_CONDITION_REPORTED");
|
||||
let encoded = serde_json::to_string(&result).expect("encoded result");
|
||||
assert!(!encoded.contains("hostname"));
|
||||
assert!(!encoded.contains("endpoint"));
|
||||
assert!(!encoded.contains("mountPath"));
|
||||
assert!(!encoded.contains("commandLine"));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn signed_health_export_binds_target_nonce_consent_and_result_digest() {
|
||||
let now = Utc::now();
|
||||
let request = request(now.timestamp());
|
||||
let identity = DeviceIdentity::generate();
|
||||
let result = evaluate_health_observation(
|
||||
&request,
|
||||
observation(now.timestamp(), 100, 50, vec!["cluster.healing"]),
|
||||
now,
|
||||
Duration::from_millis(25),
|
||||
);
|
||||
let export = sign_health_export(&request, &result, &identity, &CancellationToken::new()).expect("signed export");
|
||||
assert!(export.archive_bytes.len() < 262_144);
|
||||
|
||||
let envelope: Value = serde_json::from_slice(&export.envelope_json).expect("envelope");
|
||||
assert_eq!(envelope["organizationName"], request.organization_name);
|
||||
assert_eq!(envelope["clusterName"], request.cluster_name);
|
||||
assert_eq!(envelope["deviceName"], request.device_name);
|
||||
assert_eq!(envelope["runUid"], request.run_uid);
|
||||
assert_eq!(envelope["consentUid"], request.consent.consent_uid);
|
||||
assert_eq!(envelope["nonce"], URL_SAFE_NO_PAD.encode_to_string(request.nonce));
|
||||
assert_eq!(envelope["payload"]["sha256"], hex_lower(&Sha256::digest(&export.result_json)));
|
||||
assert_eq!(envelope["classification"], "L0");
|
||||
|
||||
let signature_document: Value = serde_json::from_slice(&export.envelope_signature).expect("signature document");
|
||||
let signature_bytes = URL_SAFE_NO_PAD
|
||||
.decode_to_vec(signature_document["value"].as_str().expect("signature value").as_bytes())
|
||||
.expect("base64 signature");
|
||||
let signature = Signature::from_slice(&signature_bytes).expect("signature");
|
||||
let verifying_key = VerifyingKey::from_public_key_der(&identity.public_key_der()).expect("public key");
|
||||
let mut signed = SIGNATURE_DOMAIN.to_vec();
|
||||
signed.extend_from_slice(&export.envelope_json);
|
||||
verifying_key.verify(&signed, &signature).expect("valid ES256 signature");
|
||||
|
||||
let mut archive = zip::ZipArchive::new(std::io::Cursor::new(export.archive_bytes)).expect("health archive");
|
||||
let mut archived_result = Vec::new();
|
||||
archive
|
||||
.by_name("result.json")
|
||||
.expect("result entry")
|
||||
.read_to_end(&mut archived_result)
|
||||
.expect("read result");
|
||||
assert_eq!(archived_result, export.result_json);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn revoked_consent_cancellation_and_oversized_results_fail_closed() {
|
||||
let now = Utc::now();
|
||||
let identity = DeviceIdentity::generate();
|
||||
let mut request = request(now.timestamp());
|
||||
let result =
|
||||
evaluate_health_observation(&request, observation(now.timestamp(), 100, 50, vec![]), now, Duration::from_millis(1));
|
||||
|
||||
request.consent.active = false;
|
||||
assert_eq!(
|
||||
sign_health_export(&request, &result, &identity, &CancellationToken::new()),
|
||||
Err(HealthError::ConsentRequired)
|
||||
);
|
||||
|
||||
request.consent.active = true;
|
||||
let cancelled = CancellationToken::new();
|
||||
cancelled.cancel();
|
||||
assert_eq!(sign_health_export(&request, &result, &identity, &cancelled), Err(HealthError::Cancelled));
|
||||
|
||||
let mut invalid_provenance = request.clone();
|
||||
invalid_provenance.provenance =
|
||||
ProfileProvenance::new("not-a-commit", "not-a-digest", "development build", vec!["unsafe feature".to_owned()]);
|
||||
assert_eq!(
|
||||
sign_health_export(&invalid_provenance, &result, &identity, &CancellationToken::new()),
|
||||
Err(HealthError::InvalidRequest)
|
||||
);
|
||||
|
||||
let oversized = evaluate_health_observation(
|
||||
&request,
|
||||
HealthSourceObservation {
|
||||
coarse_flags: Some(vec!["cluster.healing".to_owned(); 32_768]),
|
||||
..observation(now.timestamp(), 100, 50, vec![])
|
||||
},
|
||||
now,
|
||||
Duration::from_millis(1),
|
||||
);
|
||||
assert_eq!(
|
||||
sign_health_export(&request, &oversized, &identity, &CancellationToken::new()),
|
||||
Err(HealthError::LimitExceeded)
|
||||
);
|
||||
}
|
||||
|
||||
fn hex_lower(bytes: &[u8]) -> String {
|
||||
bytes.iter().fold(String::with_capacity(bytes.len() * 2), |mut value, byte| {
|
||||
use std::fmt::Write as _;
|
||||
let _ = write!(value, "{byte:02x}");
|
||||
value
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user