From 31ac243865ffced8de7a4247b8a7980e4d902c07 Mon Sep 17 00:00:00 2001 From: Chris Date: Sun, 27 Sep 2026 02:50:49 +0800 Subject: [PATCH] feat(connect): add bounded health service job --- crates/ecstore/src/disk/local.rs | 56 +- crates/ecstore/src/disk/mod.rs | 37 + .../diagnostic-scheduler/MANIFEST.sha256 | 2 + .../health-service-accept-vectors.json | 123 +++ .../health-service-reject-vectors.json | 20 + .../v1/fixtures/heartbeat/MANIFEST.sha256 | 2 +- .../heartbeat/producer-capabilities.json | 2 + protocol/agent/v1/fixtures/manifest.sha256 | 4 +- .../v1/health-service-result.schema.json | 237 +++++ protocol/agent/v1/job-envelope.schema.json | 362 ++++++++ rustfs/src/connect/diagnostics/health.rs | 827 ++++++++++++++++++ rustfs/src/connect/diagnostics/job.rs | 193 +++- rustfs/src/connect/diagnostics/mod.rs | 9 + rustfs/src/connect/diagnostics/profile_cpu.rs | 14 +- rustfs/src/connect/mod.rs | 7 + rustfs/src/storage/storage_api.rs | 2 +- rustfs/tests/connect_health.rs | 274 ++++++ 17 files changed, 2160 insertions(+), 11 deletions(-) create mode 100644 protocol/agent/v1/fixtures/diagnostic-scheduler/health-service-accept-vectors.json create mode 100644 protocol/agent/v1/fixtures/diagnostic-scheduler/health-service-reject-vectors.json create mode 100644 protocol/agent/v1/health-service-result.schema.json create mode 100644 protocol/agent/v1/job-envelope.schema.json create mode 100644 rustfs/src/connect/diagnostics/health.rs create mode 100644 rustfs/tests/connect_health.rs diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 8888d4102..077c8a90c 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -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> = + 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 { + async fn disk_info(&self, opts: &DiskInfoOptions) -> Result { + 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; } diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index 7a3f0b174..87b51dd73 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -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 diff --git a/protocol/agent/v1/fixtures/diagnostic-scheduler/MANIFEST.sha256 b/protocol/agent/v1/fixtures/diagnostic-scheduler/MANIFEST.sha256 index c74e0ed28..e9dea1049 100644 --- a/protocol/agent/v1/fixtures/diagnostic-scheduler/MANIFEST.sha256 +++ b/protocol/agent/v1/fixtures/diagnostic-scheduler/MANIFEST.sha256 @@ -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 diff --git a/protocol/agent/v1/fixtures/diagnostic-scheduler/health-service-accept-vectors.json b/protocol/agent/v1/fixtures/diagnostic-scheduler/health-service-accept-vectors.json new file mode 100644 index 000000000..eb1a2f5c9 --- /dev/null +++ b/protocol/agent/v1/fixtures/diagnostic-scheduler/health-service-accept-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" } + ] + } + } + } + ] +} diff --git a/protocol/agent/v1/fixtures/diagnostic-scheduler/health-service-reject-vectors.json b/protocol/agent/v1/fixtures/diagnostic-scheduler/health-service-reject-vectors.json new file mode 100644 index 000000000..3060f19ac --- /dev/null +++ b/protocol/agent/v1/fixtures/diagnostic-scheduler/health-service-reject-vectors.json @@ -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" } + ] +} diff --git a/protocol/agent/v1/fixtures/heartbeat/MANIFEST.sha256 b/protocol/agent/v1/fixtures/heartbeat/MANIFEST.sha256 index de8e25c91..3b8c3f8f5 100644 --- a/protocol/agent/v1/fixtures/heartbeat/MANIFEST.sha256 +++ b/protocol/agent/v1/fixtures/heartbeat/MANIFEST.sha256 @@ -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 diff --git a/protocol/agent/v1/fixtures/heartbeat/producer-capabilities.json b/protocol/agent/v1/fixtures/heartbeat/producer-capabilities.json index 27571c0fa..3220c4f99 100644 --- a/protocol/agent/v1/fixtures/heartbeat/producer-capabilities.json +++ b/protocol/agent/v1/fixtures/heartbeat/producer-capabilities.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", diff --git a/protocol/agent/v1/fixtures/manifest.sha256 b/protocol/agent/v1/fixtures/manifest.sha256 index 199ae5862..b6179b38b 100644 --- a/protocol/agent/v1/fixtures/manifest.sha256 +++ b/protocol/agent/v1/fixtures/manifest.sha256 @@ -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 diff --git a/protocol/agent/v1/health-service-result.schema.json b/protocol/agent/v1/health-service-result.schema.json new file mode 100644 index 000000000..c5ac73f0b --- /dev/null +++ b/protocol/agent/v1/health-service-result.schema.json @@ -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" } } }] } + } +} diff --git a/protocol/agent/v1/job-envelope.schema.json b/protocol/agent/v1/job-envelope.schema.json new file mode 100644 index 000000000..5f426b366 --- /dev/null +++ b/protocol/agent/v1/job-envelope.schema.json @@ -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"] + } + } + } + } + } + ] +} diff --git a/rustfs/src/connect/diagnostics/health.rs b/rustfs/src/connect/diagnostics/health.rs new file mode 100644 index 000000000..6b399ebb2 --- /dev/null +++ b/rustfs/src/connect/diagnostics/health.rs @@ -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, + pub capacity_complete: bool, + pub total_bytes: Option, + pub used_bytes: Option, + pub coarse_flags_complete: bool, + pub coarse_flags: Option>, +} + +#[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, + freshness: HealthFreshness, + complete: bool, + total_bytes: Option, + used_bytes: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct HealthFlagsObservation { + source: &'static str, + observed_at: Option, + freshness: HealthFreshness, + complete: bool, + values: Option>, +} + +#[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, + unsupported_checks: Vec, +} + +#[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, + pub envelope_signature: Vec, + pub result_json: Vec, + pub archive_bytes: Vec, + 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 { + 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 { + 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, + elapsed: std::time::Duration, +) -> HealthDiagnosticResult { + let observed_at = observation.observed_at_unix.and_then(DateTime::::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 { + 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 { + 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>, evaluated_at: DateTime, 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, 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, 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 { + DateTime::::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"); + } +} diff --git a/rustfs/src/connect/diagnostics/job.rs b/rustfs/src/connect/diagnostics/job.rs index 2db29c11f..5ebfad8fc 100644 --- a/rustfs/src/connect/diagnostics/job.rs +++ b/rustfs/src/connect/diagnostics/job.rs @@ -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, + #[serde(default, skip_serializing_if = "Option::is_none")] pub traffic_bytes: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub target_alias: Option, @@ -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 { 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 { + 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 { 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(); diff --git a/rustfs/src/connect/diagnostics/mod.rs b/rustfs/src/connect/diagnostics/mod.rs index 671e366e1..d3cf00426 100644 --- a/rustfs/src/connect/diagnostics/mod.rs +++ b/rustfs/src/connect/diagnostics/mod.rs @@ -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, diff --git a/rustfs/src/connect/diagnostics/profile_cpu.rs b/rustfs/src/connect/diagnostics/profile_cpu.rs index ac16e25bd..5e83b2376 100644 --- a/rustfs/src/connect/diagnostics/profile_cpu.rs +++ b/rustfs/src/connect/diagnostics/profile_cpu.rs @@ -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); } diff --git a/rustfs/src/connect/mod.rs b/rustfs/src/connect/mod.rs index 35be86729..614f5f9cb 100644 --- a/rustfs/src/connect/mod.rs +++ b/rustfs/src/connect/mod.rs @@ -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, diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index c0900c56c..165acfc57 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -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 { diff --git a/rustfs/tests/connect_health.rs b/rustfs/tests/connect_health.rs new file mode 100644 index 000000000..f069a3b82 --- /dev/null +++ b/rustfs/tests/connect_health.rs @@ -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, +) -> 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 + }) +}