From 95e1b84ecf51387e23851e40e33a6e35baa173ed Mon Sep 17 00:00:00 2001 From: Chris Date: Sun, 13 Sep 2026 04:33:30 +0800 Subject: [PATCH] feat(connect): add consent-bound profile exports (#7716) feat: add consent-bound profile exports --- rustfs/src/config/cli.rs | 139 ++- rustfs/src/config/mod.rs | 1 + rustfs/src/config/opt.rs | 1 + rustfs/src/connect/diagnostics/mod.rs | 11 + rustfs/src/connect/diagnostics/profile_cpu.rs | 820 ++++++++++++++++++ .../src/connect/diagnostics/profile_memory.rs | 160 ++++ .../connect/diagnostics/profile_threads.rs | 54 ++ rustfs/src/connect/mod.rs | 8 +- rustfs/src/startup_entrypoint.rs | 176 +++- rustfs/tests/connect_profile_cpu.rs | 157 ++++ rustfs/tests/connect_profile_memory.rs | 270 ++++++ rustfs/tests/connect_profile_threads.rs | 90 ++ 12 files changed, 1882 insertions(+), 5 deletions(-) create mode 100644 rustfs/src/connect/diagnostics/profile_cpu.rs create mode 100644 rustfs/src/connect/diagnostics/profile_memory.rs create mode 100644 rustfs/src/connect/diagnostics/profile_threads.rs create mode 100644 rustfs/tests/connect_profile_cpu.rs create mode 100644 rustfs/tests/connect_profile_memory.rs create mode 100644 rustfs/tests/connect_profile_threads.rs diff --git a/rustfs/src/config/cli.rs b/rustfs/src/config/cli.rs index dff87c5c4..8f9fabcfa 100644 --- a/rustfs/src/config/cli.rs +++ b/rustfs/src/config/cli.rs @@ -127,6 +127,97 @@ pub enum ConnectCommands { Register(ConnectRegisterOpts), /// Import, verify, or inspect a signed Connect service license License(ConnectLicenseOpts), + /// Capture a consent-bound local profile and write a signed export + Profile(ConnectProfileOpts), +} + +/// `connect profile` options. +#[derive(Args, Clone)] +pub struct ConnectProfileOpts { + /// Directory containing an enrolled Connect device identity + #[arg(long = "state-dir")] + pub state_dir: PathBuf, + + /// New local archive path; an existing file is never replaced + #[arg(long)] + pub output: PathBuf, + + /// Profile producer to run + #[arg(long, value_enum)] + pub tool: ConnectProfileTool, + + /// Thread source required by the threads producer + #[arg(long = "thread-scope", value_enum)] + pub thread_scope: Option, + + /// Negotiated producer schema version + #[arg(long = "schema-version", default_value_t = 1)] + pub schema_version: u16, + + /// Negotiated producer capability, such as profile.memory@1 + #[arg(long, value_parser = NonEmptyStringValueParser::new())] + pub capability: String, + + /// Organization resource name bound to the export + #[arg(long, value_parser = NonEmptyStringValueParser::new())] + pub organization: String, + + /// Cluster resource name bound to the export + #[arg(long, value_parser = NonEmptyStringValueParser::new())] + pub cluster: String, + + /// Cluster-device resource name bound to the export + #[arg(long, value_parser = NonEmptyStringValueParser::new())] + pub device: String, + + /// UUIDv7 diagnostic run identifier issued by Connect + #[arg(long = "run-uid", value_parser = NonEmptyStringValueParser::new())] + pub run_uid: String, + + /// UUIDv7 artifact identifier issued by Connect + #[arg(long = "artifact-uid", value_parser = NonEmptyStringValueParser::new())] + pub artifact_uid: String, + + /// UUIDv7 consent identifier issued by Connect + #[arg(long = "consent-uid", value_parser = NonEmptyStringValueParser::new())] + pub consent_uid: String, + + /// Consent policy revision bound to this capture + #[arg(long = "policy-revision")] + pub policy_revision: u64, + + /// Consent expiry as UTC Unix seconds + #[arg(long = "consent-expires-at")] + pub consent_expires_at_unix: i64, + + /// Artifact expiry as UTC Unix seconds + #[arg(long = "expires-at")] + pub expires_at_unix: i64, + + /// Maximum capture duration in milliseconds + #[arg(long = "duration-millis")] + pub duration_millis: u64, + + /// Sampling interval in microseconds + #[arg(long = "sample-period-micros")] + pub sample_period_micros: u64, + + /// Confirm this explicit local L3 profile capture + #[arg(long = "acknowledge-l3", required = true, action = clap::ArgAction::SetTrue)] + pub acknowledge_l3: bool, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)] +pub enum ConnectProfileTool { + Cpu, + Memory, + Threads, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)] +pub enum ConnectThreadProfileScope { + TokioRuntime, + NativeThreads, } /// `connect register` options @@ -515,6 +606,8 @@ pub enum CommandResult { ConnectRegister(ConnectRegisterOpts), /// Local Connect service-license command ConnectLicense(ConnectLicenseCommands), + /// Consent-bound local Connect profile export + ConnectProfile(ConnectProfileOpts), } /// Create default ServerOpts from environment variables @@ -651,7 +744,9 @@ mod tests { let Some(Commands::Connect(connect)) = cli.command else { panic!("connect command expected"); }; - let ConnectCommands::Register(register) = connect.command; + let ConnectCommands::Register(register) = connect.command else { + panic!("connect register command expected"); + }; assert_eq!(register.endpoint, "https://connect.example/agent/"); assert_eq!(register.ca_file, std::path::Path::new("/etc/rustfs/connect-ca.pem")); assert_eq!(register.state_dir, std::path::Path::new("/var/lib/rustfs/connect")); @@ -692,6 +787,48 @@ mod tests { assert!(help.to_string().contains("Unix only")); } + #[test] + fn connect_profile_requires_explicit_l3_acknowledgement() { + let arguments = [ + "rustfs", + "connect", + "profile", + "--state-dir", + "/var/lib/rustfs/connect", + "--output", + "/tmp/profile.zip", + "--tool", + "memory", + "--capability", + "profile.memory@1", + "--organization", + "organizations/019e3ae0-0000-7000-8000-000000000001", + "--cluster", + "organizations/019e3ae0-0000-7000-8000-000000000001/clusters/019e3ae0-0000-7000-8000-000000000002", + "--device", + "organizations/019e3ae0-0000-7000-8000-000000000001/clusters/019e3ae0-0000-7000-8000-000000000002/clusterDevices/019e3ae0-0000-7000-8000-000000000003", + "--run-uid", + "019e3ae0-0000-7000-8000-000000000004", + "--artifact-uid", + "019e3ae0-0000-7000-8000-000000000005", + "--consent-uid", + "019e3ae0-0000-7000-8000-000000000006", + "--policy-revision", + "1", + "--consent-expires-at", + "4102444800", + "--expires-at", + "4102444700", + "--duration-millis", + "10", + "--sample-period-micros", + "1000", + ]; + let error = Cli::try_parse_from(arguments).expect_err("an incomplete unacknowledged profile must fail"); + assert_eq!(error.kind(), ErrorKind::MissingRequiredArgument); + assert!(error.to_string().contains("--acknowledge-l3")); + } + #[test] fn server_help_lists_allocator_reclaim_environment() { let result = Cli::try_parse_from(["rustfs", "server", "--help"]); diff --git a/rustfs/src/config/mod.rs b/rustfs/src/config/mod.rs index ba707caf5..7fb451c36 100644 --- a/rustfs/src/config/mod.rs +++ b/rustfs/src/config/mod.rs @@ -52,6 +52,7 @@ mod config_test; // Re-export public types pub use cli::{CommandResult, InfoOpts, InfoType}; pub use cli::{ConnectLicenseArtifactOpts, ConnectLicenseCommands, ConnectLicenseScopeOpts}; +pub use cli::{ConnectProfileOpts, ConnectProfileTool, ConnectThreadProfileScope}; pub use cli::{DiagnoseFormat, DiagnoseOpts}; pub use cli::{InspectBucketMetaOpts, InspectCommands, InspectOpts}; pub use cli::{TlsCommands, TlsInspectOpts, TlsOpts}; diff --git a/rustfs/src/config/opt.rs b/rustfs/src/config/opt.rs index b552990f5..6e5fa050f 100644 --- a/rustfs/src/config/opt.rs +++ b/rustfs/src/config/opt.rs @@ -140,6 +140,7 @@ impl Opt { Some(Commands::Connect(opts)) => match opts.command { ConnectCommands::Register(opts) => Ok(CommandResult::ConnectRegister(opts)), ConnectCommands::License(opts) => Ok(CommandResult::ConnectLicense(opts.command)), + ConnectCommands::Profile(opts) => Ok(CommandResult::ConnectProfile(opts)), }, Some(Commands::Server(opts)) => Self::server_command_result(Self::from_server_opts(*opts)), None => { diff --git a/rustfs/src/connect/diagnostics/mod.rs b/rustfs/src/connect/diagnostics/mod.rs index 499a52d23..c60eafc67 100644 --- a/rustfs/src/connect/diagnostics/mod.rs +++ b/rustfs/src/connect/diagnostics/mod.rs @@ -12,8 +12,19 @@ // See the License for the specific language governing permissions and // limitations under the License. +mod profile_cpu; +mod profile_memory; +mod profile_threads; mod schedule; +pub use profile_cpu::{ + CPU_PROFILE_CAPABILITY, LocalProfileConsent, MAX_PROFILE_DURATION, MEMORY_PROFILE_CAPABILITY, PROFILE_SCHEMA_VERSION, + ProfileCaptureRequest, ProfileData, ProfileError, ProfileOutcome, ProfileProvenance, ProfileReasonCode, ProfileResult, + ProfileTool, SavedProfileExport, SignedProfileExport, THREAD_PROFILE_CAPABILITY, ThreadProfileScope, capture_cpu_profile, + encode_signed_profile_export, export_cpu_profile, save_signed_profile_export, +}; +pub use profile_memory::export_memory_profile; +pub use profile_threads::{capture_thread_profile, export_thread_profile}; pub use schedule::{ DiagnosticCollectionPolicy, DiagnosticReceipt, DiagnosticScheduleError, DiagnosticScheduleRuntime, DiagnosticScheduleStatus, ReceiptOutcome, run_local_environment_once, spawn_environment_schedule, diff --git a/rustfs/src/connect/diagnostics/profile_cpu.rs b/rustfs/src/connect/diagnostics/profile_cpu.rs new file mode 100644 index 000000000..a94f1a04e --- /dev/null +++ b/rustfs/src/connect/diagnostics/profile_cpu.rs @@ -0,0 +1,820 @@ +// 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 profile result and signed local export primitives. +//! +//! RustFS currently has no reviewed local CPU symbol catalogue or in-process +//! sampler. [`capture_cpu_profile`] therefore returns an explicit, typed +//! unsupported result. It never substitutes Pyroscope delivery state, raw +//! symbol strings, or zero samples for a local CPU measurement. + +use std::fs::{self, File, OpenOptions}; +use std::io::{Cursor, Write as _}; +use std::path::Path; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use base64_simd::URL_SAFE_NO_PAD; +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 time::{OffsetDateTime, format_description::well_known::Rfc3339}; +use tokio_util::sync::CancellationToken; +use uuid::{Uuid, Variant, Version}; +use zip::{CompressionMethod, ZipWriter, write::SimpleFileOptions}; + +use crate::connect::DeviceIdentity; + +pub const PROFILE_SCHEMA_VERSION: u16 = 1; +pub const CPU_PROFILE_CAPABILITY: &str = "profile.cpu@1"; +pub const MEMORY_PROFILE_CAPABILITY: &str = "profile.memory@1"; +pub const THREAD_PROFILE_CAPABILITY: &str = "profile.threads@1"; +pub const MAX_PROFILE_DURATION: Duration = Duration::from_secs(30); +pub const MAX_RESULT_BYTES: usize = 262_144; +pub const MAX_ENVELOPE_BYTES: usize = 16_384; +pub const MAX_ARCHIVE_BYTES: usize = 524_288; +pub const MAX_DECOMPRESSED_BYTES: usize = 278_528; +pub const MAX_BUILD_FEATURES: usize = 64; +pub const MAX_VALIDITY_SECONDS: i64 = 2_592_000; +pub const MAX_FUTURE_SKEW_SECONDS: i64 = 300; + +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 OUTPUT_MODE: u32 = 0o600; + +static PROFILE_COLLECTOR_ACTIVE: AtomicBool = AtomicBool::new(false); + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +pub enum ProfileTool { + #[serde(rename = "profile.cpu")] + Cpu, + #[serde(rename = "profile.memory")] + Memory, + #[serde(rename = "profile.threads")] + Threads, +} + +impl ProfileTool { + pub const fn id(self) -> &'static str { + match self { + Self::Cpu => "profile.cpu", + Self::Memory => "profile.memory", + Self::Threads => "profile.threads", + } + } + + pub const fn capability(self) -> &'static str { + match self { + Self::Cpu => CPU_PROFILE_CAPABILITY, + Self::Memory => MEMORY_PROFILE_CAPABILITY, + Self::Threads => THREAD_PROFILE_CAPABILITY, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum ProfileOutcome { + Succeeded, + Partial, + Failed, + Unsupported, + Cancelled, +} + +impl ProfileOutcome { + pub const fn as_str(self) -> &'static str { + match self { + Self::Succeeded => "SUCCEEDED", + Self::Partial => "PARTIAL", + Self::Failed => "FAILED", + Self::Unsupported => "UNSUPPORTED", + Self::Cancelled => "CANCELLED", + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum ProfileReasonCode { + Complete, + LimitExceeded, + SourceUnavailable, + UnsupportedTool, + UnsupportedVersion, + UnsupportedPlatform, + Cancelled, + CounterReset, + InvalidInput, + CollectionFailed, +} + +impl ProfileReasonCode { + pub const fn as_str(self) -> &'static str { + match self { + Self::Complete => "COMPLETE", + Self::LimitExceeded => "LIMIT_EXCEEDED", + Self::SourceUnavailable => "SOURCE_UNAVAILABLE", + Self::UnsupportedTool => "UNSUPPORTED_TOOL", + Self::UnsupportedVersion => "UNSUPPORTED_VERSION", + Self::UnsupportedPlatform => "UNSUPPORTED_PLATFORM", + Self::Cancelled => "CANCELLED", + Self::CounterReset => "COUNTER_RESET", + Self::InvalidInput => "INVALID_INPUT", + Self::CollectionFailed => "COLLECTION_FAILED", + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ProfileProvenance { + repository: &'static str, + source_commit: String, + executable_sha256: String, + rustfs_version: String, + os_family: ProfileOsFamily, + architecture: ProfileArchitecture, + build_features: Vec, +} + +impl ProfileProvenance { + pub fn new( + source_commit: impl Into, + executable_sha256: impl Into, + rustfs_version: impl Into, + build_features: Vec, + ) -> Self { + Self { + repository: "rustfs/rustfs", + source_commit: source_commit.into(), + executable_sha256: executable_sha256.into(), + rustfs_version: rustfs_version.into(), + os_family: ProfileOsFamily::current(), + architecture: ProfileArchitecture::current(), + build_features, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum ProfileOsFamily { + Linux, + Darwin, + Windows, + Freebsd, + Other, +} + +impl ProfileOsFamily { + fn current() -> Self { + match std::env::consts::OS { + "linux" => Self::Linux, + "macos" => Self::Darwin, + "windows" => Self::Windows, + "freebsd" => Self::Freebsd, + _ => Self::Other, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "lowercase")] +pub enum ProfileArchitecture { + #[serde(rename = "x86_64")] + X86_64, + Aarch64, + Other, +} + +impl ProfileArchitecture { + fn current() -> Self { + match std::env::consts::ARCH { + "x86_64" => Self::X86_64, + "aarch64" => Self::Aarch64, + _ => Self::Other, + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct LocalProfileConsent { + pub consent_uid: String, + pub policy_revision: u64, + pub expires_at_unix: i64, + pub confirmed: bool, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ProfileCaptureRequest { + 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: LocalProfileConsent, + pub produced_at_unix: i64, + pub expires_at_unix: i64, + pub nonce: [u8; 32], + pub duration: Duration, + pub sample_period: Duration, + pub provenance: ProfileProvenance, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ProfileCoverage { + requested_units: u32, + completed_units: u32, + unit: &'static str, +} + +impl ProfileCoverage { + pub(super) const fn complete_window() -> Self { + Self { + requested_units: 1, + completed_units: 1, + unit: "WINDOW", + } + } + + const fn none() -> Self { + Self { + requested_units: 0, + completed_units: 0, + unit: "WINDOW", + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct CpuProfileData { + sample_period_micros: u64, + samples: Vec, + dropped_sample_count: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct CpuProfileSample { + symbol_id: String, + sample_count: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct MemoryProfileData { + scope: &'static str, + allocated_bytes: u64, + allocation_count: u64, + sample_period_micros: u64, +} + +impl MemoryProfileData { + pub(super) const fn allocation_aggregates(allocated_bytes: u64, allocation_count: u64, sample_period_micros: u64) -> Self { + Self { + scope: "ALLOCATION_AGGREGATES", + allocated_bytes, + allocation_count, + sample_period_micros, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum ThreadProfileScope { + TokioRuntime, + NativeThreads, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(untagged)] +pub enum ProfileData { + Cpu(CpuProfileData), + Memory(MemoryProfileData), +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ProfileResult { + schema_version: u16, + run_uid: String, + tool_id: ProfileTool, + capability: &'static str, + outcome: ProfileOutcome, + reason_code: ProfileReasonCode, + duration_millis: u64, + provenance: ProfileProvenance, + coverage: ProfileCoverage, + data: Option, +} + +impl ProfileResult { + pub fn outcome(&self) -> ProfileOutcome { + self.outcome + } + + pub fn reason_code(&self) -> ProfileReasonCode { + self.reason_code + } + + pub fn data(&self) -> Option<&ProfileData> { + self.data.as_ref() + } + + pub(super) fn succeeded(request: &ProfileCaptureRequest, tool: ProfileTool, duration: Duration, data: ProfileData) -> Self { + Self { + schema_version: PROFILE_SCHEMA_VERSION, + run_uid: request.run_uid.clone(), + tool_id: tool, + capability: tool.capability(), + outcome: ProfileOutcome::Succeeded, + reason_code: ProfileReasonCode::Complete, + duration_millis: u64::try_from(duration.as_millis()).unwrap_or(u64::MAX).min(30_000), + provenance: request.provenance.clone(), + coverage: ProfileCoverage::complete_window(), + data: Some(data), + } + } + + pub(super) fn unsupported(request: &ProfileCaptureRequest, tool: ProfileTool, reason_code: ProfileReasonCode) -> Self { + Self { + schema_version: PROFILE_SCHEMA_VERSION, + run_uid: request.run_uid.clone(), + tool_id: tool, + capability: tool.capability(), + outcome: ProfileOutcome::Unsupported, + reason_code, + duration_millis: 0, + provenance: request.provenance.clone(), + coverage: ProfileCoverage::none(), + data: None, + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SignedProfileExport { + pub artifact_uid: String, + pub tool: ProfileTool, + pub outcome: ProfileOutcome, + pub reason_code: ProfileReasonCode, + pub archive_bytes: Vec, + pub archive_sha256: String, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SavedProfileExport { + pub artifact_uid: String, + pub archive_size_bytes: u64, + pub archive_sha256: String, +} + +#[derive(Debug, Error)] +pub enum ProfileError { + #[error("profile_local_consent_required")] + ConsentRequired, + #[error("profile_local_consent_expired")] + ConsentExpired, + #[error("profile_request_expired")] + Expired, + #[error("profile_invalid_request")] + InvalidRequest, + #[error("profile_unsupported_version")] + UnsupportedVersion, + #[error("profile_unsupported_capability")] + UnsupportedCapability, + #[error("profile_limit_exceeded")] + LimitExceeded, + #[error("profile_collection_cancelled")] + Cancelled, + #[error("profile_collection_timed_out")] + TimedOut, + #[error("profile_collection_already_running")] + Busy, + #[error("profile_source_unavailable")] + SourceUnavailable, + #[error("profile_counter_reset")] + CounterReset, + #[error("profile_export_signing_failed")] + Signing, + #[error("profile_export_exists")] + AlreadyExists, + #[error("profile_export_io_failed")] + Io(#[source] std::io::Error), + #[error("profile_export_encoding_failed")] + Encoding, + #[error("profile_export_durability_failed_after_commit")] + DurabilityAfterCommit(#[source] std::io::Error), +} + +pub fn capture_cpu_profile(request: &ProfileCaptureRequest, cancel: &CancellationToken) -> Result { + request.validate(ProfileTool::Cpu, unix_now()?)?; + check_cancel(cancel)?; + + // The only existing CPU profiler exports to Pyroscope and does not provide + // a reviewed local symbol-id catalogue. Publishing samples here would turn + // unreviewed process symbols into L3 evidence or invent their mapping. + Ok(ProfileResult::unsupported(request, ProfileTool::Cpu, ProfileReasonCode::UnsupportedTool)) +} + +pub fn export_cpu_profile( + request: &ProfileCaptureRequest, + key: &DeviceIdentity, + cancel: &CancellationToken, +) -> Result { + let result = capture_cpu_profile(request, cancel)?; + encode_signed_profile_export(request, &result, key, cancel) +} + +pub fn encode_signed_profile_export( + request: &ProfileCaptureRequest, + result: &ProfileResult, + key: &DeviceIdentity, + cancel: &CancellationToken, +) -> Result { + let now = unix_now()?; + request.validate(result.tool_id, now)?; + check_cancel(cancel)?; + let valid_data = match result.outcome { + ProfileOutcome::Succeeded | ProfileOutcome::Partial => result.data.is_some(), + ProfileOutcome::Failed | ProfileOutcome::Unsupported | ProfileOutcome::Cancelled => result.data.is_none(), + }; + if !valid_data + || result.run_uid != request.run_uid + || result.schema_version != request.schema_version + || result.capability != request.capability + { + return Err(ProfileError::InvalidRequest); + } + + let result_bytes = serde_json::to_vec(result).map_err(|_| ProfileError::Encoding)?; + if result_bytes.is_empty() || result_bytes.len() > MAX_RESULT_BYTES { + return Err(ProfileError::LimitExceeded); + } + + let device_key_id = hex_lower(&Sha256::digest(key.public_key_der())); + let envelope = ProfileEnvelope { + 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: result.tool_id, + schema_version: PROFILE_SCHEMA_VERSION, + classification: "L3", + 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: ProfilePayload { + path: RESULT_PATH, + media_type: "application/json", + size_bytes: result_bytes.len() as u64, + sha256: hex_lower(&Sha256::digest(&result_bytes)), + }, + }; + let envelope_bytes = serde_json::to_vec(&envelope).map_err(|_| ProfileError::Encoding)?; + if envelope_bytes.is_empty() || envelope_bytes.len() > MAX_ENVELOPE_BYTES { + return Err(ProfileError::LimitExceeded); + } + let signature_bytes = signature_document(key, &device_key_id, &envelope_bytes)?; + let decompressed = result_bytes + .len() + .checked_add(envelope_bytes.len()) + .and_then(|size| size.checked_add(signature_bytes.len())) + .ok_or(ProfileError::LimitExceeded)?; + if decompressed > MAX_DECOMPRESSED_BYTES { + return Err(ProfileError::LimitExceeded); + } + check_cancel(cancel)?; + if unix_now()? >= request.expires_at_unix { + return Err(ProfileError::Expired); + } + + let archive_bytes = archive(&envelope_bytes, &signature_bytes, &result_bytes)?; + if archive_bytes.len() > MAX_ARCHIVE_BYTES { + return Err(ProfileError::LimitExceeded); + } + let archive_sha256 = hex_lower(&Sha256::digest(&archive_bytes)); + + Ok(SignedProfileExport { + artifact_uid: request.artifact_uid.clone(), + tool: result.tool_id, + outcome: result.outcome, + reason_code: result.reason_code, + archive_bytes, + archive_sha256, + }) +} + +pub fn save_signed_profile_export( + output: &Path, + export: &SignedProfileExport, + cancel: &CancellationToken, +) -> Result { + check_cancel(cancel)?; + let parent = output + .parent() + .filter(|path| !path.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + let filename = output.file_name().ok_or(ProfileError::InvalidRequest)?.to_string_lossy(); + let temporary = parent.join(format!(".{filename}.{}.partial", export.artifact_uid)); + let mut options = OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt as _; + options.mode(OUTPUT_MODE); + } + + let mut file = options.open(&temporary).map_err(map_create_error)?; + let result = (|| { + file.write_all(&export.archive_bytes).map_err(ProfileError::Io)?; + check_cancel(cancel)?; + file.sync_all().map_err(ProfileError::Io)?; + check_cancel(cancel)?; + fs::hard_link(&temporary, output).map_err(map_publish_error)?; + if let Err(error) = fs::remove_file(&temporary) { + return Err(ProfileError::DurabilityAfterCommit(error)); + } + #[cfg(unix)] + if let Err(error) = File::open(parent).and_then(|directory| directory.sync_all()) { + return Err(ProfileError::DurabilityAfterCommit(error)); + } + Ok(SavedProfileExport { + artifact_uid: export.artifact_uid.clone(), + archive_size_bytes: export.archive_bytes.len() as u64, + archive_sha256: export.archive_sha256.clone(), + }) + })(); + if result.is_err() { + let _ = fs::remove_file(&temporary); + } + result +} + +impl ProfileCaptureRequest { + pub(super) fn validate(&self, tool: ProfileTool, now_unix: i64) -> Result<(), ProfileError> { + if self.schema_version != PROFILE_SCHEMA_VERSION { + return Err(ProfileError::UnsupportedVersion); + } + if self.capability != tool.capability() { + return Err(ProfileError::UnsupportedCapability); + } + if !self.consent.confirmed || self.consent.policy_revision == 0 { + return Err(ProfileError::ConsentRequired); + } + if self.consent.expires_at_unix <= now_unix || self.expires_at_unix > self.consent.expires_at_unix { + return Err(ProfileError::ConsentExpired); + } + let validity = self + .expires_at_unix + .checked_sub(self.produced_at_unix) + .ok_or(ProfileError::Expired)?; + if self.produced_at_unix > now_unix.saturating_add(MAX_FUTURE_SKEW_SECONDS) + || validity <= 0 + || self.expires_at_unix <= now_unix + || validity > MAX_VALIDITY_SECONDS + { + return Err(ProfileError::Expired); + } + if self.duration.is_zero() + || self.duration > MAX_PROFILE_DURATION + || self.sample_period.is_zero() + || self.sample_period > self.duration + { + return Err(ProfileError::LimitExceeded); + } + if !uuid7(&self.run_uid) + || !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)) + { + return Err(ProfileError::InvalidRequest); + } + Ok(()) + } +} + +pub(super) struct CollectorLease; + +impl CollectorLease { + pub(super) fn acquire() -> Result { + PROFILE_COLLECTOR_ACTIVE + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .map(|_| Self) + .map_err(|_| ProfileError::Busy) + } +} + +impl Drop for CollectorLease { + fn drop(&mut self) { + PROFILE_COLLECTOR_ACTIVE.store(false, Ordering::Release); + } +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct ProfileEnvelope<'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: ProfileTool, + 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: ProfilePayload, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct ProfilePayload { + path: &'static str, + media_type: &'static str, + size_bytes: u64, + sha256: String, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct ProfileSignature<'a> { + algorithm: &'static str, + key_id: &'a str, + value: String, +} + +fn signature_document(key: &DeviceIdentity, key_id: &str, envelope: &[u8]) -> Result, ProfileError> { + let pkcs8 = key.to_pkcs8_der().map_err(|_| ProfileError::Signing)?; + let signing_key = SigningKey::from_pkcs8_der(pkcs8.as_slice()).map_err(|_| ProfileError::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); + let value = URL_SAFE_NO_PAD.encode_to_string(signature.normalize_s().to_bytes()); + serde_json::to_vec(&ProfileSignature { + algorithm: "ES256", + key_id, + value, + }) + .map_err(|_| ProfileError::Encoding) +} + +fn archive(envelope: &[u8], signature: &[u8], result: &[u8]) -> Result, ProfileError> { + let cursor = Cursor::new(Vec::with_capacity(envelope.len() + signature.len() + result.len() + 512)); + let mut writer = ZipWriter::new(cursor); + let options = SimpleFileOptions::DEFAULT + .compression_method(CompressionMethod::Stored) + .unix_permissions(OUTPUT_MODE); + for (name, bytes) in [(ENVELOPE_PATH, envelope), (SIGNATURE_PATH, signature), (RESULT_PATH, result)] { + writer.start_file(name, options).map_err(|_| ProfileError::Encoding)?; + writer.write_all(bytes).map_err(ProfileError::Io)?; + } + writer + .finish() + .map(|cursor| cursor.into_inner()) + .map_err(|_| ProfileError::Encoding) +} + +fn resource_names_match(request: &ProfileCaptureRequest) -> bool { + let Some(organization_uid) = request.organization_name.strip_prefix("organizations/") else { + return false; + }; + if !uuid7(organization_uid) { + return false; + } + let cluster_prefix = format!("{}/clusters/", request.organization_name); + let Some(cluster_uid) = request.cluster_name.strip_prefix(&cluster_prefix) else { + return false; + }; + if !uuid7(cluster_uid) { + return false; + } + let device_prefix = format!("{}/clusterDevices/", request.cluster_name); + request.device_name.strip_prefix(&device_prefix).is_some_and(uuid7) +} + +fn uuid7(value: &str) -> bool { + Uuid::parse_str(value).is_ok_and(|uuid| { + uuid.get_version() == Some(Version::SortRand) && uuid.get_variant() == Variant::RFC4122 && uuid.to_string() == value + }) +} + +fn lower_hex(value: &str, length: usize) -> bool { + value.len() == length + && value + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) +} + +fn build_feature(value: &str) -> bool { + value.len() <= 64 + && value.as_bytes().first().is_some_and(u8::is_ascii_lowercase) + && value + .bytes() + .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'_' | b'-')) +} + +fn version(value: &str) -> bool { + if value.is_empty() + || value.len() > 64 + || !value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'-')) + { + return false; + } + let (core, suffix) = value + .split_once('-') + .map_or((value, None), |(core, suffix)| (core, Some(suffix))); + if suffix.is_some_and(str::is_empty) { + return false; + } + let mut parts = core.split('.'); + parts.clone().count() == 3 && parts.all(|part| !part.is_empty() && part.bytes().all(|byte| byte.is_ascii_digit())) +} + +fn timestamp(unix: i64) -> Result { + OffsetDateTime::from_unix_timestamp(unix) + .map_err(|_| ProfileError::InvalidRequest)? + .format(&Rfc3339) + .map_err(|_| ProfileError::InvalidRequest) +} + +pub(super) fn unix_now() -> Result { + let duration = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|_| ProfileError::InvalidRequest)?; + i64::try_from(duration.as_secs()).map_err(|_| ProfileError::InvalidRequest) +} + +pub(super) fn check_cancel(cancel: &CancellationToken) -> Result<(), ProfileError> { + if cancel.is_cancelled() { + Err(ProfileError::Cancelled) + } else { + Ok(()) + } +} + +fn hex_lower(bytes: &[u8]) -> String { + let mut value = String::with_capacity(bytes.len() * 2); + for byte in bytes { + use std::fmt::Write as _; + write!(&mut value, "{byte:02x}").expect("writing hexadecimal to a string cannot fail"); + } + value +} + +fn map_create_error(error: std::io::Error) -> ProfileError { + if error.kind() == std::io::ErrorKind::AlreadyExists { + ProfileError::AlreadyExists + } else { + ProfileError::Io(error) + } +} + +fn map_publish_error(error: std::io::Error) -> ProfileError { + if error.kind() == std::io::ErrorKind::AlreadyExists { + ProfileError::AlreadyExists + } else { + ProfileError::Io(error) + } +} diff --git a/rustfs/src/connect/diagnostics/profile_memory.rs b/rustfs/src/connect/diagnostics/profile_memory.rs new file mode 100644 index 000000000..ba77f83da --- /dev/null +++ b/rustfs/src/connect/diagnostics/profile_memory.rs @@ -0,0 +1,160 @@ +// 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. + +//! Mimalloc allocation aggregates for the bounded profile contract. +//! +//! Only cumulative allocated octets and allocation counts are sampled. Heap +//! bytes, addresses, stack traces, paths, symbols, and allocator debug text are +//! excluded from the result. + +use std::future::Future; +use std::pin::Pin; +use std::time::{Duration, Instant}; + +use serde_json::Value; +use tokio_util::sync::CancellationToken; + +use crate::connect::DeviceIdentity; + +use super::profile_cpu::{ + CollectorLease, MemoryProfileData, ProfileCaptureRequest, ProfileData, ProfileError, ProfileResult, ProfileTool, + SignedProfileExport, check_cancel, encode_signed_profile_export, unix_now, +}; + +const MAX_ALLOCATOR_STATS_BYTES: usize = 262_144; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) struct AllocationSnapshot { + total_allocated_bytes: u64, + allocation_count: u64, +} + +pub(crate) trait AllocationProfileSource: Send + Sync { + fn snapshot(&self) -> Result; +} + +struct MimallocProfileSource; + +impl AllocationProfileSource for MimallocProfileSource { + fn snapshot(&self) -> Result { + #[cfg(target_os = "windows")] + { + Err(ProfileError::SourceUnavailable) + } + #[cfg(not(target_os = "windows"))] + { + parse_allocator_stats(&rustfs_mimalloc::MiMalloc::stats_json()) + } + } +} + +pub async fn export_memory_profile( + request: &ProfileCaptureRequest, + key: &DeviceIdentity, + cancel: &CancellationToken, +) -> Result { + export_memory_profile_from(request, key, cancel, &MimallocProfileSource).await +} + +pub(crate) async fn export_memory_profile_from( + request: &ProfileCaptureRequest, + key: &DeviceIdentity, + cancel: &CancellationToken, + source: &dyn AllocationProfileSource, +) -> Result { + request.validate(ProfileTool::Memory, unix_now()?)?; + check_cancel(cancel)?; + let _lease = CollectorLease::acquire()?; + let started = Instant::now(); + let result = collect_window(request.sample_period, cancel, source); + let (before, after) = tokio::time::timeout(request.duration, result) + .await + .map_err(|_| ProfileError::TimedOut)??; + check_cancel(cancel)?; + + let allocated_bytes = after + .total_allocated_bytes + .checked_sub(before.total_allocated_bytes) + .ok_or(ProfileError::CounterReset)?; + let allocation_count = after + .allocation_count + .checked_sub(before.allocation_count) + .ok_or(ProfileError::CounterReset)?; + let elapsed = started.elapsed(); + let sample_period_micros = u64::try_from(elapsed.as_micros()).map_err(|_| ProfileError::LimitExceeded)?; + let data = MemoryProfileData::allocation_aggregates(allocated_bytes, allocation_count, sample_period_micros.max(1)); + let result = ProfileResult::succeeded(request, ProfileTool::Memory, elapsed, ProfileData::Memory(data)); + + encode_signed_profile_export(request, &result, key, cancel) +} + +fn collect_window<'a>( + sample_period: Duration, + cancel: &'a CancellationToken, + source: &'a dyn AllocationProfileSource, +) -> Pin> + Send + 'a>> { + Box::pin(async move { + let before = source.snapshot()?; + tokio::select! { + () = cancel.cancelled() => return Err(ProfileError::Cancelled), + () = tokio::time::sleep(sample_period) => {} + } + let after = source.snapshot()?; + Ok((before, after)) + }) +} + +pub(crate) fn parse_allocator_stats(stats: &str) -> Result { + if stats.is_empty() || stats.len() > MAX_ALLOCATOR_STATS_BYTES { + return Err(ProfileError::SourceUnavailable); + } + let value = serde_json::from_str::(stats).map_err(|_| ProfileError::SourceUnavailable)?; + let total_allocated_bytes = sum_metrics(&value, &["malloc_normal", "malloc_huge"], "total")?; + let allocation_count = sum_metrics(&value, &["malloc_normal_count", "malloc_huge_count"], "total")?; + Ok(AllocationSnapshot { + total_allocated_bytes, + allocation_count, + }) +} + +fn sum_metrics(value: &Value, metrics: &[&str], field: &str) -> Result { + metrics.iter().try_fold(0_u64, |sum, metric| { + let value = metric_field(value, metric, field).ok_or(ProfileError::SourceUnavailable)?; + sum.checked_add(value).ok_or(ProfileError::LimitExceeded) + }) +} + +fn metric_field(value: &Value, metric: &str, field: &str) -> Option { + match value { + Value::Object(fields) => { + if let Some(metric_value) = fields.get(metric) + && let Some(value) = numeric_field(metric_value, field) + { + return Some(value); + } + fields.values().find_map(|value| metric_field(value, metric, field)) + } + Value::Array(values) => values.iter().find_map(|value| metric_field(value, metric, field)), + _ => None, + } +} + +fn numeric_field(value: &Value, field: &str) -> Option { + match value { + Value::Number(number) => number.as_u64(), + Value::String(value) => value.parse().ok(), + Value::Object(fields) => fields.get(field).and_then(|value| numeric_field(value, field)), + _ => None, + } +} diff --git a/rustfs/src/connect/diagnostics/profile_threads.rs b/rustfs/src/connect/diagnostics/profile_threads.rs new file mode 100644 index 000000000..1bf405755 --- /dev/null +++ b/rustfs/src/connect/diagnostics/profile_threads.rs @@ -0,0 +1,54 @@ +// 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. + +//! Explicit thread/runtime profile capability result. +//! +//! Dial9 currently exposes session and disk-buffer state, not bounded counts +//! for RUNNABLE, WAITING, BLOCKED, and UNKNOWN threads. Native thread state +//! would require a separate reviewed platform adapter. Neither source is +//! relabelled as the contract's thread data. + +use tokio_util::sync::CancellationToken; + +use super::profile_cpu::{ + ProfileCaptureRequest, ProfileError, ProfileReasonCode, ProfileResult, ProfileTool, ThreadProfileScope, check_cancel, + encode_signed_profile_export, unix_now, +}; +use crate::connect::DeviceIdentity; + +use super::profile_cpu::SignedProfileExport; + +pub fn capture_thread_profile( + request: &ProfileCaptureRequest, + _scope: ThreadProfileScope, + cancel: &CancellationToken, +) -> Result { + request.validate(ProfileTool::Threads, unix_now()?)?; + check_cancel(cancel)?; + Ok(ProfileResult::unsupported( + request, + ProfileTool::Threads, + ProfileReasonCode::UnsupportedTool, + )) +} + +pub fn export_thread_profile( + request: &ProfileCaptureRequest, + scope: ThreadProfileScope, + key: &DeviceIdentity, + cancel: &CancellationToken, +) -> Result { + let result = capture_thread_profile(request, scope, cancel)?; + encode_signed_profile_export(request, &result, key, cancel) +} diff --git a/rustfs/src/connect/mod.rs b/rustfs/src/connect/mod.rs index 1c1689ffd..2f8d07b06 100644 --- a/rustfs/src/connect/mod.rs +++ b/rustfs/src/connect/mod.rs @@ -45,8 +45,12 @@ pub use client::{ClientError, ConnectClient, ConnectConfig}; pub use config::{HeartbeatConfig, HeartbeatConfigError, HeartbeatSchedule}; pub use credential_store::{CredentialStore, DeviceCredential}; pub use diagnostics::{ - DiagnosticCollectionPolicy, DiagnosticReceipt, DiagnosticScheduleError, DiagnosticScheduleRuntime, DiagnosticScheduleStatus, - ReceiptOutcome, run_local_environment_once, spawn_environment_schedule, + CPU_PROFILE_CAPABILITY, DiagnosticCollectionPolicy, DiagnosticReceipt, DiagnosticScheduleError, DiagnosticScheduleRuntime, + DiagnosticScheduleStatus, LocalProfileConsent, MAX_PROFILE_DURATION, MEMORY_PROFILE_CAPABILITY, PROFILE_SCHEMA_VERSION, + ProfileCaptureRequest, ProfileData, ProfileError, ProfileOutcome, ProfileProvenance, ProfileReasonCode, ProfileResult, + ProfileTool, ReceiptOutcome, SavedProfileExport, SignedProfileExport, THREAD_PROFILE_CAPABILITY, ThreadProfileScope, + capture_cpu_profile, capture_thread_profile, encode_signed_profile_export, export_cpu_profile, export_memory_profile, + export_thread_profile, run_local_environment_once, save_signed_profile_export, spawn_environment_schedule, }; pub use environment::{ ENVIRONMENT_CAPABILITY, ENVIRONMENT_SCHEMA_VERSION, EnvironmentCollectionRequest, EnvironmentError, diff --git a/rustfs/src/startup_entrypoint.rs b/rustfs/src/startup_entrypoint.rs index 37db46013..bff08dd12 100644 --- a/rustfs/src/startup_entrypoint.rs +++ b/rustfs/src/startup_entrypoint.rs @@ -13,7 +13,10 @@ // limitations under the License. use crate::{ - config::{CommandResult, Config, ConnectLicenseCommands, ConnectLicenseScopeOpts, Opt}, + config::{ + CommandResult, Config, ConnectLicenseCommands, ConnectLicenseScopeOpts, ConnectProfileOpts, ConnectProfileTool, + ConnectThreadProfileScope, Opt, + }, startup_lifecycle::{StartupRuntimeLifecycle, run_startup_runtime_lifecycle}, startup_preflight::{StartupServerPreflightError, bootstrap_external_prefix_compat, init_startup_server_preflight}, startup_server::{StartupHttpServers, StartupListenContext, init_startup_http_servers, init_startup_listen_context}, @@ -22,7 +25,8 @@ use crate::{ storage_api::server::http::ServerContextSlot, storage_api::startup::storage::bootstrap_instance_ctx, }; -use std::io::{Error, Result}; +use std::io::{Error, Read as _, Result}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tracing::{error, instrument}; const LOG_COMPONENT_MAIN: &str = "main"; @@ -130,6 +134,7 @@ async fn async_main() -> Result<()> { return Ok(()); } CommandResult::ConnectLicense(command) => return execute_connect_license(command), + CommandResult::ConnectProfile(options) => return execute_connect_profile(options).await, CommandResult::Server(config) => config, }; @@ -159,6 +164,173 @@ async fn async_main() -> Result<()> { } } +async fn execute_connect_profile(options: ConnectProfileOpts) -> Result<()> { + use crate::connect::{ + IdentityStore, LocalProfileConsent, ProfileCaptureRequest, ProfileProvenance, ThreadProfileScope, export_cpu_profile, + export_memory_profile, export_thread_profile, save_signed_profile_export, + }; + use rand::{TryRng as _, rngs::SysRng}; + + let key = IdentityStore::new(options.state_dir.join("identity")) + .load() + .map_err(Error::other)? + .ok_or_else(|| Error::other("connect profile requires an enrolled device identity"))?; + let executable_sha256 = hash_current_executable()?; + let produced_at_unix = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(Error::other) + .and_then(|duration| i64::try_from(duration.as_secs()).map_err(Error::other))?; + let mut nonce = [0_u8; 32]; + SysRng.try_fill_bytes(&mut nonce).map_err(Error::other)?; + let request = ProfileCaptureRequest { + organization_name: options.organization, + cluster_name: options.cluster, + device_name: options.device, + run_uid: options.run_uid, + artifact_uid: options.artifact_uid, + schema_version: options.schema_version, + capability: options.capability, + consent: LocalProfileConsent { + consent_uid: options.consent_uid, + policy_revision: options.policy_revision, + expires_at_unix: options.consent_expires_at_unix, + confirmed: options.acknowledge_l3, + }, + produced_at_unix, + expires_at_unix: options.expires_at_unix, + nonce, + duration: Duration::from_millis(options.duration_millis), + sample_period: Duration::from_micros(options.sample_period_micros), + provenance: ProfileProvenance::new( + crate::version::build::COMMIT_HASH, + executable_sha256, + env!("CARGO_PKG_VERSION"), + enabled_build_features(), + ), + }; + let cancel = tokio_util::sync::CancellationToken::new(); + let capture = async { + match options.tool { + ConnectProfileTool::Cpu => { + if options.thread_scope.is_some() { + return Err(Error::other("--thread-scope is valid only for the threads profile")); + } + export_cpu_profile(&request, &key, &cancel).map_err(Error::other) + } + ConnectProfileTool::Memory => { + if options.thread_scope.is_some() { + return Err(Error::other("--thread-scope is valid only for the threads profile")); + } + export_memory_profile(&request, &key, &cancel).await.map_err(Error::other) + } + ConnectProfileTool::Threads => { + let scope = match options.thread_scope { + Some(ConnectThreadProfileScope::TokioRuntime) => ThreadProfileScope::TokioRuntime, + Some(ConnectThreadProfileScope::NativeThreads) => ThreadProfileScope::NativeThreads, + None => return Err(Error::other("--thread-scope is required for the threads profile")), + }; + export_thread_profile(&request, scope, &key, &cancel).map_err(Error::other) + } + } + }; + tokio::pin!(capture); + let export = tokio::select! { + biased; + signal = tokio::signal::ctrl_c() => { + signal.map_err(Error::other)?; + cancel.cancel(); + return Err(Error::other("profile collection cancelled")); + } + result = capture.as_mut() => result?, + }; + drop(capture); + let tool = export.tool; + let outcome = export.outcome; + let reason_code = export.reason_code; + let output = options.output; + let writer_cancel = cancel.clone(); + let mut writer = tokio::task::spawn_blocking(move || save_signed_profile_export(&output, &export, &writer_cancel)); + let receipt = tokio::select! { + biased; + signal = tokio::signal::ctrl_c() => { + signal.map_err(Error::other)?; + cancel.cancel(); + writer.await.map_err(Error::other)?.map_err(Error::other)? + } + result = &mut writer => result.map_err(Error::other)?.map_err(Error::other)?, + }; + + println!("tool={} outcome={} reason={}", tool.id(), outcome.as_str(), reason_code.as_str()); + println!( + "artifact={} bytes={} sha256={}", + receipt.artifact_uid, receipt.archive_size_bytes, receipt.archive_sha256 + ); + println!("upload=not-performed"); + Ok(()) +} + +fn hash_current_executable() -> Result { + use sha2::{Digest as _, Sha256}; + + const MAX_EXECUTABLE_BYTES: u64 = 1_073_741_824; + let path = std::env::current_exe().map_err(Error::other)?; + let mut file = std::fs::File::open(path).map_err(Error::other)?; + let metadata = file.metadata().map_err(Error::other)?; + if !metadata.is_file() || metadata.len() == 0 || metadata.len() > MAX_EXECUTABLE_BYTES { + return Err(Error::other("current executable is outside the profile provenance limit")); + } + let mut hasher = Sha256::new(); + let mut buffer = [0_u8; 64 * 1024]; + let mut read_bytes = 0_u64; + loop { + let count = file.read(&mut buffer).map_err(Error::other)?; + if count == 0 { + break; + } + read_bytes = read_bytes + .checked_add(u64::try_from(count).map_err(Error::other)?) + .ok_or_else(|| Error::other("current executable is outside the profile provenance limit"))?; + if read_bytes > MAX_EXECUTABLE_BYTES { + return Err(Error::other("current executable is outside the profile provenance limit")); + } + hasher.update(&buffer[..count]); + } + if read_bytes != metadata.len() { + return Err(Error::other("current executable changed while hashing profile provenance")); + } + Ok(hex_simd::encode_to_string(hasher.finalize(), hex_simd::AsciiCase::Lower)) +} + +fn enabled_build_features() -> Vec { + let mut features = Vec::new(); + for (enabled, name) in [ + (cfg!(feature = "connect-e2e-short-credentials"), "connect-e2e-short-credentials"), + (cfg!(feature = "dial9"), "dial9"), + (cfg!(feature = "e2e-test-hooks"), "e2e-test-hooks"), + (cfg!(feature = "ftps"), "ftps"), + (cfg!(feature = "full"), "full"), + (cfg!(feature = "gcs"), "gcs"), + (cfg!(feature = "hotpath"), "hotpath"), + (cfg!(feature = "hotpath-alloc"), "hotpath-alloc"), + (cfg!(feature = "hotpath-cpu"), "hotpath-cpu"), + (cfg!(feature = "io-scheduler-debug"), "io-scheduler-debug"), + (cfg!(feature = "license"), "license"), + (cfg!(feature = "metrics-gpu"), "metrics-gpu"), + (cfg!(feature = "offline-enrollment-e2e-root"), "offline-enrollment-e2e-root"), + (cfg!(feature = "pyroscope"), "pyroscope"), + (cfg!(feature = "rio-v2"), "rio-v2"), + (cfg!(feature = "sftp"), "sftp"), + (cfg!(feature = "swift"), "swift"), + (cfg!(feature = "tracing-chunk-debug"), "tracing-chunk-debug"), + (cfg!(feature = "webdav"), "webdav"), + ] { + if enabled { + features.push(name.to_owned()); + } + } + features +} + fn execute_connect_license(command: ConnectLicenseCommands) -> Result<()> { use crate::connect::{apply_license_artifact, inspect_installed_license, verify_license_artifact}; diff --git a/rustfs/tests/connect_profile_cpu.rs b/rustfs/tests/connect_profile_cpu.rs new file mode 100644 index 000000000..06537d0a3 --- /dev/null +++ b/rustfs/tests/connect_profile_cpu.rs @@ -0,0 +1,157 @@ +// 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. + +mod connect { + pub use rustfs::connect::DeviceIdentity; +} + +#[allow(dead_code)] +#[path = "../src/connect/diagnostics/profile_cpu.rs"] +mod profile_cpu; + +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use profile_cpu::{ + CPU_PROFILE_CAPABILITY, LocalProfileConsent, MAX_PROFILE_DURATION, ProfileCaptureRequest, ProfileError, ProfileOutcome, + ProfileProvenance, ProfileReasonCode, capture_cpu_profile, +}; +use tokio_util::sync::CancellationToken; + +fn now() -> i64 { + SystemTime::now().duration_since(UNIX_EPOCH).expect("current time").as_secs() as i64 +} + +fn request() -> ProfileCaptureRequest { + let now = now(); + let organization = "organizations/019e3ae0-0000-7000-8000-000000000001"; + let cluster = format!("{organization}/clusters/019e3ae0-0000-7000-8000-000000000002"); + ProfileCaptureRequest { + organization_name: organization.to_string(), + cluster_name: cluster.clone(), + device_name: format!("{cluster}/clusterDevices/019e3ae0-0000-7000-8000-000000000003"), + run_uid: "019e3ae0-0000-7000-8000-000000000004".to_string(), + artifact_uid: "019e3ae0-0000-7000-8000-000000000005".to_string(), + schema_version: 1, + capability: CPU_PROFILE_CAPABILITY.to_string(), + consent: LocalProfileConsent { + consent_uid: "019e3ae0-0000-7000-8000-000000000006".to_string(), + policy_revision: 1, + expires_at_unix: now + 120, + confirmed: true, + }, + produced_at_unix: now, + expires_at_unix: now + 60, + nonce: [0x5a; 32], + duration: Duration::from_secs(1), + sample_period: Duration::from_millis(10), + provenance: ProfileProvenance::new("a".repeat(40), "b".repeat(64), "1.0.0-rc.6", vec!["default".to_string()]), + } +} + +#[test] +fn cpu_without_reviewed_symbol_catalog_is_explicitly_unsupported() { + let result = capture_cpu_profile(&request(), &CancellationToken::new()).expect("unsupported is a typed result"); + assert_eq!(result.outcome(), ProfileOutcome::Unsupported); + assert_eq!(result.reason_code(), ProfileReasonCode::UnsupportedTool); + assert!(result.data().is_none(), "unsupported CPU must not fabricate zero samples"); + + let json = serde_json::to_value(result).expect("result JSON"); + assert_eq!(json["toolId"], "profile.cpu"); + assert_eq!(json["capability"], "profile.cpu@1"); + assert_eq!(json["coverage"]["requestedUnits"], 0); + assert_eq!(json["coverage"]["completedUnits"], 0); + assert!(json["data"].is_null()); +} + +#[test] +fn cpu_refuses_missing_or_expired_local_consent_before_capability_disclosure() { + let mut missing = request(); + missing.consent.confirmed = false; + assert!(matches!( + capture_cpu_profile(&missing, &CancellationToken::new()), + Err(ProfileError::ConsentRequired) + )); + + let mut expired = request(); + expired.consent.expires_at_unix = now() - 1; + assert!(matches!( + capture_cpu_profile(&expired, &CancellationToken::new()), + Err(ProfileError::ConsentExpired) + )); +} + +#[test] +fn cpu_negotiation_and_resource_limits_are_closed_at_the_boundary() { + let mut invalid = request(); + invalid.schema_version = 2; + assert!(matches!( + capture_cpu_profile(&invalid, &CancellationToken::new()), + Err(ProfileError::UnsupportedVersion) + )); + + let mut invalid = request(); + invalid.capability = "profile.cpu@2".to_string(); + assert!(matches!( + capture_cpu_profile(&invalid, &CancellationToken::new()), + Err(ProfileError::UnsupportedCapability) + )); + + let mut boundary = request(); + boundary.duration = MAX_PROFILE_DURATION; + boundary.sample_period = MAX_PROFILE_DURATION; + assert!(capture_cpu_profile(&boundary, &CancellationToken::new()).is_ok()); + + let mut over = request(); + over.duration = MAX_PROFILE_DURATION + Duration::from_nanos(1); + assert!(matches!( + capture_cpu_profile(&over, &CancellationToken::new()), + Err(ProfileError::LimitExceeded) + )); + + let mut features = request(); + features.provenance = ProfileProvenance::new( + "a".repeat(40), + "b".repeat(64), + "1.0.0", + (0..65).map(|index| format!("feature_{index}")).collect(), + ); + assert!(matches!( + capture_cpu_profile(&features, &CancellationToken::new()), + Err(ProfileError::InvalidRequest) + )); + + let mut empty_version_suffix = request(); + empty_version_suffix.provenance = + ProfileProvenance::new("a".repeat(40), "b".repeat(64), "1.0.0-", vec!["default".to_string()]); + assert!(matches!( + capture_cpu_profile(&empty_version_suffix, &CancellationToken::new()), + Err(ProfileError::InvalidRequest) + )); + + let mut overflowing_window = request(); + overflowing_window.produced_at_unix = i64::MIN; + overflowing_window.expires_at_unix = i64::MAX; + overflowing_window.consent.expires_at_unix = i64::MAX; + assert!(matches!( + capture_cpu_profile(&overflowing_window, &CancellationToken::new()), + Err(ProfileError::Expired) + )); +} + +#[test] +fn cpu_honors_pre_cancelled_capture() { + let cancel = CancellationToken::new(); + cancel.cancel(); + assert!(matches!(capture_cpu_profile(&request(), &cancel), Err(ProfileError::Cancelled))); +} diff --git a/rustfs/tests/connect_profile_memory.rs b/rustfs/tests/connect_profile_memory.rs new file mode 100644 index 000000000..d66069df7 --- /dev/null +++ b/rustfs/tests/connect_profile_memory.rs @@ -0,0 +1,270 @@ +// 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. + +mod connect { + pub use rustfs::connect::DeviceIdentity; +} + +#[allow(dead_code)] +#[path = "../src/connect/diagnostics/profile_cpu.rs"] +mod profile_cpu; +#[path = "../src/connect/diagnostics/profile_memory.rs"] +mod profile_memory; + +use std::fs; +use std::io::{Cursor, Read as _}; +#[cfg(unix)] +use std::os::unix::fs::PermissionsExt as _; +use std::sync::Mutex; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use base64_simd::URL_SAFE_NO_PAD; +use p256::ecdsa::signature::Verifier as _; +use p256::ecdsa::{Signature, VerifyingKey}; +use p256::pkcs8::DecodePublicKey as _; +use profile_cpu::{ + LocalProfileConsent, MEMORY_PROFILE_CAPABILITY, ProfileCaptureRequest, ProfileError, ProfileProvenance, + save_signed_profile_export, +}; +use profile_memory::{AllocationProfileSource, export_memory_profile, export_memory_profile_from, parse_allocator_stats}; +use sha2::{Digest as _, Sha256}; +use tokio_util::sync::CancellationToken; +use zip::ZipArchive; + +static TEST_PROFILE_LOCK: Mutex<()> = Mutex::new(()); + +fn profile_test_lock() -> std::sync::MutexGuard<'static, ()> { + TEST_PROFILE_LOCK.lock().unwrap_or_else(std::sync::PoisonError::into_inner) +} + +#[global_allocator] +static GLOBAL: rustfs_mimalloc::MiMalloc = rustfs_mimalloc::MiMalloc; + +fn now() -> i64 { + SystemTime::now().duration_since(UNIX_EPOCH).expect("current time").as_secs() as i64 +} + +fn request() -> ProfileCaptureRequest { + let now = now(); + let organization = "organizations/019e3ae0-0000-7000-8000-000000000011"; + let cluster = format!("{organization}/clusters/019e3ae0-0000-7000-8000-000000000012"); + ProfileCaptureRequest { + organization_name: organization.to_string(), + cluster_name: cluster.clone(), + device_name: format!("{cluster}/clusterDevices/019e3ae0-0000-7000-8000-000000000013"), + run_uid: "019e3ae0-0000-7000-8000-000000000014".to_string(), + artifact_uid: "019e3ae0-0000-7000-8000-000000000015".to_string(), + schema_version: 1, + capability: MEMORY_PROFILE_CAPABILITY.to_string(), + consent: LocalProfileConsent { + consent_uid: "019e3ae0-0000-7000-8000-000000000016".to_string(), + policy_revision: 7, + expires_at_unix: now + 120, + confirmed: true, + }, + produced_at_unix: now, + expires_at_unix: now + 60, + nonce: [0x6b; 32], + duration: Duration::from_secs(1), + sample_period: Duration::from_millis(10), + provenance: ProfileProvenance::new("c".repeat(40), "d".repeat(64), "1.0.0-rc.6", vec!["default".to_string()]), + } +} + +struct SequenceSource { + values: Mutex>, +} + +impl SequenceSource { + fn new(first: &'static str, second: &'static str) -> Self { + Self { + values: Mutex::new(vec![second, first]), + } + } +} + +impl AllocationProfileSource for SequenceSource { + fn snapshot(&self) -> Result { + let value = self.values.lock().expect("sequence lock").pop().expect("two samples"); + parse_allocator_stats(value) + } +} + +fn stats(bytes: u64, count: u64) -> String { + format!( + r#"{{"malloc_normal":{{"total":{bytes}}},"malloc_huge":{{"total":0}},"malloc_normal_count":{{"total":{count}}},"malloc_huge_count":{{"total":0}}}}"# + ) +} + +fn archive_entry(archive: &mut ZipArchive>>, name: &str) -> Vec { + let mut entry = archive.by_name(name).expect("archive entry"); + let mut bytes = Vec::new(); + entry.read_to_end(&mut bytes).expect("read archive entry"); + bytes +} + +#[tokio::test] +async fn real_mimalloc_profile_produces_a_signed_three_file_export() { + let _guard = profile_test_lock(); + let key = connect::DeviceIdentity::generate(); + let export = export_memory_profile(&request(), &key, &CancellationToken::new()) + .await + .expect("real allocation aggregate export"); + assert!(export.archive_bytes.len() <= profile_cpu::MAX_ARCHIVE_BYTES); + assert_eq!(export.archive_sha256, hex(&Sha256::digest(&export.archive_bytes))); + + let mut archive = ZipArchive::new(Cursor::new(export.archive_bytes.clone())).expect("profile archive"); + assert_eq!(archive.len(), 3); + let envelope_bytes = archive_entry(&mut archive, "envelope.json"); + let signature_bytes = archive_entry(&mut archive, "envelope.sig"); + let result_bytes = archive_entry(&mut archive, "result.json"); + let envelope: serde_json::Value = serde_json::from_slice(&envelope_bytes).expect("envelope JSON"); + let signature: serde_json::Value = serde_json::from_slice(&signature_bytes).expect("signature JSON"); + let result: serde_json::Value = serde_json::from_slice(&result_bytes).expect("result JSON"); + + assert_eq!(envelope["classification"], "L3"); + assert_eq!(envelope["payload"]["path"], "result.json"); + assert_eq!(envelope["payload"]["sizeBytes"], result_bytes.len()); + assert_eq!(envelope["payload"]["sha256"], hex(&Sha256::digest(&result_bytes))); + assert_eq!(result["toolId"], "profile.memory"); + assert_eq!(result["outcome"], "SUCCEEDED"); + assert_eq!(result["data"]["scope"], "ALLOCATION_AGGREGATES"); + assert!(result["data"]["allocatedBytes"].is_u64()); + assert!(result["data"]["allocationCount"].is_u64()); + assert!(result["data"]["samplePeriodMicros"].as_u64().is_some_and(|value| value > 0)); + let encoded = serde_json::to_string(&result).expect("encoded result"); + for forbidden in ["/Users/", "/proc/", "AKIA", "secret", "stack", "symbol"] { + assert!(!encoded.contains(forbidden), "result leaked forbidden material: {forbidden}"); + } + + assert_eq!(signature["algorithm"], "ES256"); + let raw_signature = URL_SAFE_NO_PAD + .decode_to_vec(signature["value"].as_str().expect("signature value")) + .expect("base64url signature"); + assert_eq!(raw_signature.len(), 64); + let signature_value = Signature::from_slice(&raw_signature).expect("P-256 signature"); + assert_eq!(signature_value.normalize_s(), signature_value); + let public = VerifyingKey::from_public_key_der(&key.public_key_der()).expect("public key"); + let mut input = b"rustfs-diagnostic-envelope-v1\0".to_vec(); + input.extend_from_slice(&envelope_bytes); + public + .verify(&input, &signature_value) + .expect("signature over exact envelope bytes"); +} + +#[tokio::test] +async fn memory_profile_reports_counter_reset_and_cancellation_without_an_artifact() { + let _guard = profile_test_lock(); + let first = Box::leak(stats(100, 10).into_boxed_str()); + let second = Box::leak(stats(90, 11).into_boxed_str()); + let source = SequenceSource::new(first, second); + let key = connect::DeviceIdentity::generate(); + assert!(matches!( + export_memory_profile_from(&request(), &key, &CancellationToken::new(), &source).await, + Err(ProfileError::CounterReset) + )); + + let first = Box::leak(stats(100, 10).into_boxed_str()); + let second = Box::leak(stats(120, 12).into_boxed_str()); + let source = SequenceSource::new(first, second); + let mut request = request(); + request.sample_period = Duration::from_secs(1); + request.duration = Duration::from_secs(2); + let cancel = CancellationToken::new(); + let cancellation = async { + tokio::time::sleep(Duration::from_millis(5)).await; + cancel.cancel(); + }; + let (result, ()) = tokio::join!(export_memory_profile_from(&request, &key, &cancel, &source), cancellation); + assert!(matches!(result, Err(ProfileError::Cancelled))); +} + +#[tokio::test] +async fn memory_profile_uses_only_bounded_allocator_aggregates() { + let _guard = profile_test_lock(); + let first = Box::leak(stats(1_000, 20).into_boxed_str()); + let second = Box::leak(stats(1_250, 24).into_boxed_str()); + let source = SequenceSource::new(first, second); + let key = connect::DeviceIdentity::generate(); + let export = export_memory_profile_from(&request(), &key, &CancellationToken::new(), &source) + .await + .expect("aggregate export"); + let mut archive = ZipArchive::new(Cursor::new(export.archive_bytes)).expect("profile archive"); + let result: serde_json::Value = serde_json::from_slice(&archive_entry(&mut archive, "result.json")).expect("result JSON"); + assert_eq!(result["data"]["allocatedBytes"], 250); + assert_eq!(result["data"]["allocationCount"], 4); + + let oversized = format!("{}{}", stats(1, 1), " ".repeat(262_145)); + assert!(matches!(parse_allocator_stats(&oversized), Err(ProfileError::SourceUnavailable))); +} + +#[tokio::test] +async fn memory_profile_allows_only_one_collector_at_a_time() { + let _guard = profile_test_lock(); + let first = Box::leak(stats(100, 10).into_boxed_str()); + let second = Box::leak(stats(120, 12).into_boxed_str()); + let source = SequenceSource::new(first, second); + let second_source = SequenceSource::new(first, second); + let key = connect::DeviceIdentity::generate(); + let mut request = request(); + request.sample_period = Duration::from_millis(30); + + let first_cancel = CancellationToken::new(); + let second_cancel = CancellationToken::new(); + let first_capture = export_memory_profile_from(&request, &key, &first_cancel, &source); + let second_capture = async { + tokio::time::sleep(Duration::from_millis(5)).await; + export_memory_profile_from(&request, &key, &second_cancel, &second_source).await + }; + let (first_result, second_result) = tokio::join!(first_capture, second_capture); + assert!(first_result.is_ok()); + assert!(matches!(second_result, Err(ProfileError::Busy))); +} + +#[tokio::test] +async fn signed_export_is_private_no_clobber_and_cancel_safe() { + let _guard = profile_test_lock(); + let first = Box::leak(stats(100, 10).into_boxed_str()); + let second = Box::leak(stats(150, 12).into_boxed_str()); + let source = SequenceSource::new(first, second); + let key = connect::DeviceIdentity::generate(); + let export = export_memory_profile_from(&request(), &key, &CancellationToken::new(), &source) + .await + .expect("aggregate export"); + let directory = tempfile::tempdir().expect("temporary output"); + let output = directory.path().join("profile.zip"); + let receipt = save_signed_profile_export(&output, &export, &CancellationToken::new()).expect("save export"); + assert_eq!(receipt.archive_sha256, export.archive_sha256); + assert_eq!(receipt.archive_size_bytes, export.archive_bytes.len() as u64); + assert_eq!(fs::read(&output).expect("saved archive"), export.archive_bytes); + #[cfg(unix)] + assert_eq!(fs::metadata(&output).expect("metadata").permissions().mode() & 0o777, 0o600); + assert!(matches!( + save_signed_profile_export(&output, &export, &CancellationToken::new()), + Err(ProfileError::AlreadyExists) + )); + + let cancelled_output = directory.path().join("cancelled.zip"); + let cancelled = CancellationToken::new(); + cancelled.cancel(); + assert!(matches!( + save_signed_profile_export(&cancelled_output, &export, &cancelled), + Err(ProfileError::Cancelled) + )); + assert!(!cancelled_output.exists()); +} + +fn hex(bytes: &[u8]) -> String { + bytes.iter().map(|byte| format!("{byte:02x}")).collect() +} diff --git a/rustfs/tests/connect_profile_threads.rs b/rustfs/tests/connect_profile_threads.rs new file mode 100644 index 000000000..da2938e05 --- /dev/null +++ b/rustfs/tests/connect_profile_threads.rs @@ -0,0 +1,90 @@ +// 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. + +mod connect { + pub use rustfs::connect::DeviceIdentity; +} + +#[allow(dead_code)] +#[path = "../src/connect/diagnostics/profile_cpu.rs"] +mod profile_cpu; +#[path = "../src/connect/diagnostics/profile_threads.rs"] +mod profile_threads; + +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use profile_cpu::{ + LocalProfileConsent, ProfileCaptureRequest, ProfileError, ProfileOutcome, ProfileProvenance, ProfileReasonCode, + THREAD_PROFILE_CAPABILITY, ThreadProfileScope, +}; +use profile_threads::capture_thread_profile; +use tokio_util::sync::CancellationToken; + +fn request() -> ProfileCaptureRequest { + let now = SystemTime::now().duration_since(UNIX_EPOCH).expect("current time").as_secs() as i64; + let organization = "organizations/019e3ae0-0000-7000-8000-000000000021"; + let cluster = format!("{organization}/clusters/019e3ae0-0000-7000-8000-000000000022"); + ProfileCaptureRequest { + organization_name: organization.to_string(), + cluster_name: cluster.clone(), + device_name: format!("{cluster}/clusterDevices/019e3ae0-0000-7000-8000-000000000023"), + run_uid: "019e3ae0-0000-7000-8000-000000000024".to_string(), + artifact_uid: "019e3ae0-0000-7000-8000-000000000025".to_string(), + schema_version: 1, + capability: THREAD_PROFILE_CAPABILITY.to_string(), + consent: LocalProfileConsent { + consent_uid: "019e3ae0-0000-7000-8000-000000000026".to_string(), + policy_revision: 1, + expires_at_unix: now + 120, + confirmed: true, + }, + produced_at_unix: now, + expires_at_unix: now + 60, + nonce: [0x7c; 32], + duration: Duration::from_secs(1), + sample_period: Duration::from_millis(10), + provenance: ProfileProvenance::new("e".repeat(40), "f".repeat(64), "1.0.0-rc.6", vec!["default".to_string()]), + } +} + +#[test] +fn tokio_and_native_thread_scopes_remain_explicitly_unsupported() { + for scope in [ThreadProfileScope::TokioRuntime, ThreadProfileScope::NativeThreads] { + let result = capture_thread_profile(&request(), scope, &CancellationToken::new()).expect("unsupported result"); + assert_eq!(result.outcome(), ProfileOutcome::Unsupported); + assert_eq!(result.reason_code(), ProfileReasonCode::UnsupportedTool); + assert!(result.data().is_none(), "unsupported scope must not publish zero state counts"); + let json = serde_json::to_value(result).expect("result JSON"); + assert_eq!(json["toolId"], "profile.threads"); + assert_eq!(json["capability"], "profile.threads@1"); + assert!(json["data"].is_null()); + } +} + +#[test] +fn thread_profile_rejects_wrong_negotiation_and_cancellation() { + let mut invalid = request(); + invalid.capability = "profile.threads@2".to_string(); + assert!(matches!( + capture_thread_profile(&invalid, ThreadProfileScope::TokioRuntime, &CancellationToken::new()), + Err(ProfileError::UnsupportedCapability) + )); + + let cancel = CancellationToken::new(); + cancel.cancel(); + assert!(matches!( + capture_thread_profile(&request(), ThreadProfileScope::NativeThreads, &cancel), + Err(ProfileError::Cancelled) + )); +}