From f2cdca42eddd44681cd220d7b9dfdbbe91c719a1 Mon Sep 17 00:00:00 2001 From: Chris Date: Sun, 13 Sep 2026 07:00:08 +0800 Subject: [PATCH] Add bounded local telemetry export commands (#7720) feat(connect): add bounded telemetry producers --- Cargo.lock | 2 + rustfs/Cargo.toml | 2 + rustfs/src/config/cli.rs | 146 +++ rustfs/src/config/mod.rs | 4 + rustfs/src/config/opt.rs | 1 + rustfs/src/connect/diagnostics/mod.rs | 18 + .../src/connect/diagnostics/trace_analysis.rs | 149 +++ rustfs/src/connect/diagnostics/trace_otlp.rs | 271 +++++ .../src/connect/diagnostics/trace_record.rs | 1062 +++++++++++++++++ .../src/connect/diagnostics/trace_replay.rs | 127 ++ rustfs/src/connect/mod.rs | 23 +- rustfs/src/startup_entrypoint.rs | 232 +++- rustfs/tests/connect_trace_analysis.rs | 46 + rustfs/tests/connect_trace_otlp.rs | 220 ++++ rustfs/tests/connect_trace_record.rs | 440 +++++++ rustfs/tests/connect_trace_replay.rs | 135 +++ 16 files changed, 2869 insertions(+), 9 deletions(-) create mode 100644 rustfs/src/connect/diagnostics/trace_analysis.rs create mode 100644 rustfs/src/connect/diagnostics/trace_otlp.rs create mode 100644 rustfs/src/connect/diagnostics/trace_record.rs create mode 100644 rustfs/src/connect/diagnostics/trace_replay.rs create mode 100644 rustfs/tests/connect_trace_analysis.rs create mode 100644 rustfs/tests/connect_trace_otlp.rs create mode 100644 rustfs/tests/connect_trace_record.rs create mode 100644 rustfs/tests/connect_trace_replay.rs diff --git a/Cargo.lock b/Cargo.lock index 96ac766f1..77c06869f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9526,12 +9526,14 @@ dependencies = [ "mime_guess", "moka", "opentelemetry", + "opentelemetry-proto", "opentelemetry_sdk", "p256 0.14.0", "parking_lot", "percent-encoding", "pin-project-lite", "proptest", + "prost 0.14.4", "quick-xml", "rand 0.10.2", "rcgen", diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index a7bb3b7ea..2fa7e3247 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -307,6 +307,7 @@ bytes = { workspace = true, features = ["serde"] } chrono = { workspace = true, features = ["serde"] } flate2 = { workspace = true } flatbuffers.workspace = true +prost.workspace = true rmp-serde.workspace = true quick-xml.workspace = true rustfs-signer.workspace = true @@ -364,6 +365,7 @@ chacha20poly1305 = { workspace = true } # Observability and Metrics metrics = { workspace = true } opentelemetry = { workspace = true } +opentelemetry-proto = { workspace = true, features = ["trace"] } tracing-opentelemetry = { workspace = true } # Data structures hashbrown = { workspace = true, features = ["serde", "rayon"] } diff --git a/rustfs/src/config/cli.rs b/rustfs/src/config/cli.rs index a969b0914..f77643e40 100644 --- a/rustfs/src/config/cli.rs +++ b/rustfs/src/config/cli.rs @@ -131,6 +131,113 @@ pub enum ConnectCommands { Profile(ConnectProfileOpts), /// Capture allow-listed local log events and write a signed export Logs(ConnectLogsOpts), + /// Record, forward, or replay consent-bound telemetry + Telemetry(ConnectTelemetryOpts), +} + +#[derive(Args, Clone)] +pub struct ConnectTelemetryOpts { + #[command(subcommand)] + pub command: ConnectTelemetryCommands, +} + +#[derive(Subcommand, Clone)] +pub enum ConnectTelemetryCommands { + /// Capture process-local telemetry when an approved typed source is available + Record(ConnectTelemetryRecordOpts), + /// Forward an OTLP protobuf batch from stdin to a customer-approved collector + Otlp(ConnectTelemetryOtlpOpts), + /// Replay reviewed trace JSON read from stdin and write a signed export + Replay(ConnectTelemetryReplayOpts), +} + +#[derive(Args, Clone)] +pub struct ConnectTelemetryArtifactOpts { + /// 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, + + /// 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 operation + #[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, + + /// Confirm this explicit local L3 telemetry operation + #[arg(long = "acknowledge-l3", required = true, action = clap::ArgAction::SetTrue)] + pub acknowledge_l3: bool, +} + +#[derive(Args, Clone)] +pub struct ConnectTelemetryRecordOpts { + #[command(flatten)] + pub artifact: ConnectTelemetryArtifactOpts, + + /// Capture duration in milliseconds + #[arg(long = "duration-millis")] + pub duration_millis: u64, + + /// Maximum exported span count + #[arg(long = "max-spans", default_value_t = 1_024)] + pub max_spans: usize, +} + +#[derive(Args, Clone)] +pub struct ConnectTelemetryOtlpOpts { + #[command(flatten)] + pub artifact: ConnectTelemetryArtifactOpts, + + /// Customer-approved OTLP/HTTP traces endpoint + #[arg(long, value_parser = NonEmptyStringValueParser::new())] + pub endpoint: String, + + /// Forward timeout in milliseconds + #[arg(long = "timeout-millis")] + pub timeout_millis: u64, + + /// Environment variable containing the local Authorization header value + #[arg(long = "authorization-env", value_parser = NonEmptyStringValueParser::new())] + pub authorization_env: Option, +} + +#[derive(Args, Clone)] +pub struct ConnectTelemetryReplayOpts { + #[command(flatten)] + pub artifact: ConnectTelemetryArtifactOpts, } /// `connect logs` options. @@ -690,6 +797,8 @@ pub enum CommandResult { ConnectProfile(ConnectProfileOpts), /// Consent-bound local Connect log export ConnectLogs(ConnectLogsOpts), + /// Consent-bound local Connect telemetry operation + ConnectTelemetry(ConnectTelemetryCommands), } /// Create default ServerOpts from environment variables @@ -947,6 +1056,43 @@ mod tests { assert!(error.to_string().contains("--acknowledge-l3")); } + #[test] + fn connect_telemetry_record_requires_explicit_l3_acknowledgement() { + let arguments = [ + "rustfs", + "connect", + "telemetry", + "record", + "--state-dir", + "/var/lib/rustfs/connect", + "--output", + "/tmp/telemetry.zip", + "--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", + ]; + let error = Cli::try_parse_from(arguments).expect_err("unacknowledged telemetry capture 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 e340642dd..469cbdf0d 100644 --- a/rustfs/src/config/mod.rs +++ b/rustfs/src/config/mod.rs @@ -54,6 +54,10 @@ pub use cli::{CommandResult, InfoOpts, InfoType}; pub use cli::{ConnectLicenseArtifactOpts, ConnectLicenseCommands, ConnectLicenseScopeOpts}; pub use cli::{ConnectLogsMode, ConnectLogsOpts}; pub use cli::{ConnectProfileOpts, ConnectProfileTool, ConnectThreadProfileScope}; +pub use cli::{ + ConnectTelemetryArtifactOpts, ConnectTelemetryCommands, ConnectTelemetryOtlpOpts, ConnectTelemetryRecordOpts, + ConnectTelemetryReplayOpts, +}; 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 8387f7c43..d2ca22b3b 100644 --- a/rustfs/src/config/opt.rs +++ b/rustfs/src/config/opt.rs @@ -142,6 +142,7 @@ impl Opt { ConnectCommands::License(opts) => Ok(CommandResult::ConnectLicense(opts.command)), ConnectCommands::Profile(opts) => Ok(CommandResult::ConnectProfile(opts)), ConnectCommands::Logs(opts) => Ok(CommandResult::ConnectLogs(opts)), + ConnectCommands::Telemetry(opts) => Ok(CommandResult::ConnectTelemetry(opts.command)), }, 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 278cef299..4860d0c4f 100644 --- a/rustfs/src/connect/diagnostics/mod.rs +++ b/rustfs/src/connect/diagnostics/mod.rs @@ -17,6 +17,10 @@ mod profile_cpu; mod profile_memory; mod profile_threads; mod schedule; +mod trace_analysis; +mod trace_otlp; +mod trace_record; +mod trace_replay; pub use logs::{ CaptureMode, LOGS_CAPABILITY, LOGS_SCHEMA_VERSION, LocalLogConsent, LogCaptureError, LogCaptureRequest, LogProvenance, @@ -34,3 +38,17 @@ pub use schedule::{ DiagnosticCollectionPolicy, DiagnosticReceipt, DiagnosticScheduleError, DiagnosticScheduleRuntime, DiagnosticScheduleStatus, ReceiptOutcome, run_local_environment_once, spawn_environment_schedule, }; +pub use trace_analysis::{OperationSummary, TraceAnalysis, TraceAnalysisError, analyze_trace}; +pub use trace_otlp::{ + LocalOtlpHeaders, MAX_OTLP_BODY_BYTES, OtlpBatch, OtlpForwardError, OtlpReceipt, export_trace_otlp, export_trace_otlp_result, +}; +pub use trace_record::{ + LocalTelemetryConsent, MAX_SAFE_INTEGER, MAX_TELEMETRY_DURATION, MAX_TELEMETRY_RESULT_BYTES, MAX_TELEMETRY_SPANS, + ObservedTelemetrySpan, RecordedTrace, SavedTelemetryExport, SignedTelemetryExport, TELEMETRY_OTLP_CAPABILITY, + TELEMETRY_RECORD_CAPABILITY, TELEMETRY_REPLAY_CAPABILITY, TELEMETRY_SCHEMA_VERSION, TelemetryArtifactConsent, + TelemetryArtifactError, TelemetryArtifactRequest, TelemetryCoverage, TelemetryDiagnosticResult, TelemetryOperation, + TelemetryOutcome, TelemetryProducerError, TelemetryProvenance, TelemetryReasonCode, TelemetrySpan, TelemetrySpanStatus, + TelemetryTool, TraceRecordCapture, TraceRecordCompletion, TraceRecordLimits, encode_signed_telemetry_export, + record_diagnostic_result, record_trace, record_trace_bus, save_signed_telemetry_export, +}; +pub use trace_replay::{LocallyReviewedTraceArtifact, ReplayedTrace, TraceReplayError, replay_trace, replay_trace_result}; diff --git a/rustfs/src/connect/diagnostics/trace_analysis.rs b/rustfs/src/connect/diagnostics/trace_analysis.rs new file mode 100644 index 000000000..8db444d04 --- /dev/null +++ b/rustfs/src/connect/diagnostics/trace_analysis.rs @@ -0,0 +1,149 @@ +// 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. + +//! Deterministic local summaries over redacted telemetry spans. + +use serde::{Deserialize, Serialize}; +use thiserror::Error; +use tokio_util::sync::CancellationToken; + +use super::trace_record::{ + LocalTelemetryConsent, TelemetryOperation, TelemetryProducerError, TelemetrySpanStatus, acquire_telemetry_lease, +}; +use super::trace_replay::ReplayedTrace; + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub struct TraceAnalysis { + pub input_artifact_sha256: String, + pub span_count: u64, + pub error_count: u64, + pub total_duration_micros: u64, + pub operations: Vec, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub struct OperationSummary { + pub operation: TelemetryOperation, + pub span_count: u64, + pub error_count: u64, + pub total_duration_micros: u64, +} + +#[derive(Debug, Error, Eq, PartialEq)] +pub enum TraceAnalysisError { + #[error("another telemetry operation is already running")] + Busy, + #[error("local telemetry consent is expired")] + ConsentExpired, + #[error("telemetry analysis was cancelled")] + Cancelled, + #[error("telemetry analysis counters overflowed")] + CounterOverflow, +} + +pub fn analyze_trace( + replay: &ReplayedTrace, + consent: LocalTelemetryConsent, + cancel: &CancellationToken, +) -> Result { + consent.remaining().map_err(map_consent_error)?; + if cancel.is_cancelled() { + return Err(TraceAnalysisError::Cancelled); + } + let _lease = acquire_telemetry_lease().map_err(map_consent_error)?; + + let mut operations = [ + OperationSummary { + operation: TelemetryOperation::GetObject, + span_count: 0, + error_count: 0, + total_duration_micros: 0, + }, + OperationSummary { + operation: TelemetryOperation::PutObject, + span_count: 0, + error_count: 0, + total_duration_micros: 0, + }, + OperationSummary { + operation: TelemetryOperation::HeadObject, + span_count: 0, + error_count: 0, + total_duration_micros: 0, + }, + OperationSummary { + operation: TelemetryOperation::ListObjects, + span_count: 0, + error_count: 0, + total_duration_micros: 0, + }, + OperationSummary { + operation: TelemetryOperation::InternalRpc, + span_count: 0, + error_count: 0, + total_duration_micros: 0, + }, + ]; + let mut error_count = 0u64; + let mut total_duration_micros = 0u64; + for span in &replay.spans { + if cancel.is_cancelled() { + return Err(TraceAnalysisError::Cancelled); + } + let summary = &mut operations[operation_index(span.operation)]; + summary.span_count = summary.span_count.checked_add(1).ok_or(TraceAnalysisError::CounterOverflow)?; + summary.total_duration_micros = summary + .total_duration_micros + .checked_add(span.duration_micros) + .ok_or(TraceAnalysisError::CounterOverflow)?; + total_duration_micros = total_duration_micros + .checked_add(span.duration_micros) + .ok_or(TraceAnalysisError::CounterOverflow)?; + if span.status == TelemetrySpanStatus::Error { + summary.error_count = summary + .error_count + .checked_add(1) + .ok_or(TraceAnalysisError::CounterOverflow)?; + error_count = error_count.checked_add(1).ok_or(TraceAnalysisError::CounterOverflow)?; + } + } + let span_count = u64::try_from(replay.spans.len()).map_err(|_| TraceAnalysisError::CounterOverflow)?; + Ok(TraceAnalysis { + input_artifact_sha256: replay.input_artifact_sha256.clone(), + span_count, + error_count, + total_duration_micros, + operations: operations.into_iter().filter(|summary| summary.span_count != 0).collect(), + }) +} + +const fn operation_index(operation: TelemetryOperation) -> usize { + match operation { + TelemetryOperation::GetObject => 0, + TelemetryOperation::PutObject => 1, + TelemetryOperation::HeadObject => 2, + TelemetryOperation::ListObjects => 3, + TelemetryOperation::InternalRpc => 4, + } +} + +fn map_consent_error(error: TelemetryProducerError) -> TraceAnalysisError { + match error { + TelemetryProducerError::Busy => TraceAnalysisError::Busy, + TelemetryProducerError::ConsentExpired => TraceAnalysisError::ConsentExpired, + _ => TraceAnalysisError::CounterOverflow, + } +} diff --git a/rustfs/src/connect/diagnostics/trace_otlp.rs b/rustfs/src/connect/diagnostics/trace_otlp.rs new file mode 100644 index 000000000..510310b87 --- /dev/null +++ b/rustfs/src/connect/diagnostics/trace_otlp.rs @@ -0,0 +1,271 @@ +// 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. + +//! Customer-side OTLP/HTTP forwarding with bounded payloads and local secrets. + +use std::time::{Duration, Instant}; + +use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest; +use prost::Message as _; +use reqwest::{ + Client, StatusCode, Url, + header::{self, HeaderMap}, +}; +use serde::{Deserialize, Serialize}; +use thiserror::Error; +use tokio_util::sync::CancellationToken; + +use super::trace_record::{ + LocalTelemetryConsent, MAX_TELEMETRY_DURATION, TelemetryArtifactRequest, TelemetryDiagnosticResult, TelemetryLease, + TelemetryProducerError, TelemetryTool, acquire_telemetry_lease, ensure_result_size, +}; + +pub const MAX_OTLP_BODY_BYTES: usize = 1_048_576; + +/// An encoded OTLP/HTTP protobuf batch from the local OpenTelemetry adapter. +/// The bytes are never included in the diagnostic result or error values. +pub struct OtlpBatch { + body: Vec, + span_count: u64, + _lease: TelemetryLease, +} + +impl OtlpBatch { + pub fn new(body: Vec) -> Result { + if body.is_empty() || body.len() > MAX_OTLP_BODY_BYTES { + return Err(OtlpForwardError::InvalidBatch); + } + let lease = acquire_telemetry_lease().map_err(map_producer_error)?; + let request = ExportTraceServiceRequest::decode(body.as_slice()).map_err(|_| OtlpForwardError::InvalidBatch)?; + if request.resource_spans.len() > 1024 { + return Err(OtlpForwardError::InvalidBatch); + } + let scope_count = request.resource_spans.iter().try_fold(0_usize, |count, resource| { + count + .checked_add(resource.scope_spans.len()) + .filter(|count| *count <= 1024) + .ok_or(OtlpForwardError::InvalidBatch) + })?; + if scope_count == 0 { + return Err(OtlpForwardError::InvalidBatch); + } + let span_count = request + .resource_spans + .iter() + .flat_map(|resource| &resource.scope_spans) + .try_fold(0_u64, |count, scope| { + let spans = u64::try_from(scope.spans.len()).map_err(|_| OtlpForwardError::InvalidBatch)?; + count.checked_add(spans).ok_or(OtlpForwardError::InvalidBatch) + })?; + if span_count == 0 || span_count > 1024 { + return Err(OtlpForwardError::InvalidBatch); + } + Ok(Self { + body, + span_count, + _lease: lease, + }) + } +} + +pub async fn export_trace_otlp_result( + request: &TelemetryArtifactRequest, + endpoint: Url, + headers: LocalOtlpHeaders, + batch: OtlpBatch, + consent: LocalTelemetryConsent, + timeout: Duration, + cancel: &CancellationToken, +) -> Result, OtlpForwardError> { + let started = Instant::now(); + let receipt = export_trace_otlp(endpoint, headers, batch, consent, timeout, cancel).await?; + Ok(TelemetryDiagnosticResult::succeeded( + request, + TelemetryTool::Otlp, + started.elapsed(), + receipt, + )) +} + +/// Locally supplied collector authentication. This type has no `Debug` or +/// serialization implementation, keeping credentials out of results and logs. +pub struct LocalOtlpHeaders(HeaderMap); + +impl LocalOtlpHeaders { + pub fn new(headers: HeaderMap) -> Self { + Self(headers) + } +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub struct OtlpReceipt { + pub accepted_span_count: u64, + pub rejected_span_count: u64, + pub exported_bytes: u64, +} + +#[derive(Debug, Error, Eq, PartialEq)] +pub enum OtlpForwardError { + #[error("another telemetry operation is already running")] + Busy, + #[error("OTLP endpoint must be HTTPS or an HTTP loopback address without credentials, query or fragment")] + Endpoint, + #[error("OTLP batch must contain 1..1024 spans and 1..1048576 bytes")] + InvalidBatch, + #[error("OTLP forward timeout must be between 1ms and 30s")] + InvalidTimeout, + #[error("OTLP forwarding was cancelled")] + Cancelled, + #[error("local OTLP collector rejected the batch")] + Rejected, + #[error("local OTLP collector is unavailable")] + Unavailable, + #[error("OTLP receipt exceeds its result limit")] + ResultTooLarge, + #[error("local telemetry consent is expired")] + ConsentExpired, +} + +pub async fn export_trace_otlp( + endpoint: Url, + headers: LocalOtlpHeaders, + batch: OtlpBatch, + consent: LocalTelemetryConsent, + timeout: Duration, + cancel: &CancellationToken, +) -> Result { + validate_endpoint(&endpoint)?; + if timeout.is_zero() || timeout > MAX_TELEMETRY_DURATION { + return Err(OtlpForwardError::InvalidTimeout); + } + let remaining = consent.remaining().map_err(map_consent_error)?; + let consent_binds = remaining <= timeout; + let timeout = timeout.min(remaining); + let timeout_millis = usize::try_from(timeout.as_millis()).map_err(|_| OtlpForwardError::InvalidTimeout)?; + if timeout_millis == 0 { + return Err(OtlpForwardError::InvalidTimeout); + } + let deadline = tokio::time::Instant::now() + timeout; + let OtlpBatch { + body, + span_count, + _lease, + } = batch; + let byte_budget = MAX_OTLP_BODY_BYTES + .checked_mul(timeout_millis.min(1000)) + .and_then(|bytes| bytes.checked_div(1000)) + .ok_or(OtlpForwardError::InvalidBatch)?; + if body.len() > byte_budget { + return Err(OtlpForwardError::InvalidBatch); + } + let exported_bytes = u64::try_from(body.len()).map_err(|_| OtlpForwardError::InvalidBatch)?; + let mut headers = headers.0; + for transport_header in [ + header::HOST, + header::CONTENT_LENGTH, + header::CONTENT_TYPE, + header::CONNECTION, + header::TRANSFER_ENCODING, + ] { + headers.remove(transport_header); + } + let client = Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .timeout(timeout) + .build() + .map_err(|_| OtlpForwardError::Unavailable)?; + + let request = client + .post(endpoint) + .headers(headers) + .header(reqwest::header::CONTENT_TYPE, "application/x-protobuf") + .body(body); + let mut response = tokio::select! { + biased; + _ = cancel.cancelled() => return Err(OtlpForwardError::Cancelled), + _ = tokio::time::sleep_until(deadline) => return Err(deadline_error(consent_binds)), + response = request.send() => response.map_err(|_| OtlpForwardError::Unavailable)?, + }; + if response.status() != StatusCode::OK { + return Err(if response.status().is_client_error() { + OtlpForwardError::Rejected + } else { + OtlpForwardError::Unavailable + }); + } + loop { + let chunk = tokio::select! { + biased; + _ = cancel.cancelled() => return Err(OtlpForwardError::Cancelled), + _ = tokio::time::sleep_until(deadline) => return Err(deadline_error(consent_binds)), + chunk = response.chunk() => chunk.map_err(|_| OtlpForwardError::Unavailable)?, + }; + let Some(chunk) = chunk else { + break; + }; + if !chunk.is_empty() { + // A successful OTLP response may carry partial-success details. + // Until the approved protobuf adapter is wired, refusing such a + // response avoids claiming every span was accepted. + return Err(OtlpForwardError::Rejected); + } + } + + let receipt = OtlpReceipt { + accepted_span_count: span_count, + rejected_span_count: 0, + exported_bytes, + }; + ensure_result_size(&receipt).map_err(|_| OtlpForwardError::ResultTooLarge)?; + Ok(receipt) +} + +fn deadline_error(consent_binds: bool) -> OtlpForwardError { + if consent_binds { + OtlpForwardError::ConsentExpired + } else { + OtlpForwardError::Unavailable + } +} + +fn validate_endpoint(endpoint: &Url) -> Result<(), OtlpForwardError> { + let loopback_http = endpoint.scheme() == "http" + && endpoint + .host_str() + .is_some_and(|host| matches!(host, "localhost" | "127.0.0.1" | "[::1]" | "::1")); + if (endpoint.scheme() != "https" && !loopback_http) + || !endpoint.username().is_empty() + || endpoint.password().is_some() + || endpoint.query().is_some() + || endpoint.fragment().is_some() + { + return Err(OtlpForwardError::Endpoint); + } + Ok(()) +} + +fn map_consent_error(error: TelemetryProducerError) -> OtlpForwardError { + match error { + TelemetryProducerError::ConsentExpired => OtlpForwardError::ConsentExpired, + _ => OtlpForwardError::Unavailable, + } +} + +fn map_producer_error(error: TelemetryProducerError) -> OtlpForwardError { + match error { + TelemetryProducerError::Busy => OtlpForwardError::Busy, + _ => OtlpForwardError::InvalidBatch, + } +} diff --git a/rustfs/src/connect/diagnostics/trace_record.rs b/rustfs/src/connect/diagnostics/trace_record.rs new file mode 100644 index 000000000..eeb382b9f --- /dev/null +++ b/rustfs/src/connect/diagnostics/trace_record.rs @@ -0,0 +1,1062 @@ +// 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, locally-authorized capture of already classified trace spans. +//! +//! The producer accepts only the five operations frozen by the Connect +//! diagnostic contract. Request paths, bucket/object names, HTTP headers and +//! arbitrary attributes cannot enter the captured representation. + +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, Instant, SystemTime, UNIX_EPOCH}; + +use base64_simd::URL_SAFE_NO_PAD; +use p256::ecdsa::{Signature, SigningKey, signature::Signer as _}; +use p256::pkcs8::DecodePrivateKey as _; +use rustfs_common::trace_bus::subscribe_trace_events; +use serde::{Deserialize, Serialize}; +use sha2::{Digest as _, Sha256}; +use thiserror::Error; +use time::{OffsetDateTime, format_description::well_known::Rfc3339}; +use tokio::sync::mpsc; +use tokio_util::sync::CancellationToken; +use uuid::{Uuid, Variant, Version}; +use zip::{CompressionMethod, ZipWriter, write::SimpleFileOptions}; + +use crate::connect::DeviceIdentity; + +pub const MAX_TELEMETRY_DURATION: Duration = Duration::from_secs(30); +pub const MAX_TELEMETRY_SPANS: usize = 1024; +pub const MAX_TELEMETRY_RESULT_BYTES: usize = 262_144; +pub const MAX_SAFE_INTEGER: u64 = 9_007_199_254_740_991; +pub const TELEMETRY_SCHEMA_VERSION: u16 = 1; +pub const TELEMETRY_RECORD_CAPABILITY: &str = "telemetry.record@1"; +pub const TELEMETRY_OTLP_CAPABILITY: &str = "telemetry.otlp@1"; +pub const TELEMETRY_REPLAY_CAPABILITY: &str = "telemetry.replay@1"; + +const SIGNATURE_DOMAIN: &[u8] = b"rustfs-diagnostic-envelope-v1\0"; +const ENVELOPE_PATH: &str = "envelope.json"; +const SIGNATURE_PATH: &str = "envelope.sig"; +const RESULT_PATH: &str = "result.json"; +const MAX_ARCHIVE_BYTES: usize = 524_288; +const MAX_ENVELOPE_BYTES: usize = 16_384; +const MAX_DECOMPRESSED_BYTES: usize = 278_528; +const MAX_VALIDITY_SECONDS: i64 = 2_592_000; +const MAX_FUTURE_SKEW_SECONDS: i64 = 300; +const MAX_BUILD_FEATURES: usize = 64; +static TELEMETRY_LEASED: AtomicBool = AtomicBool::new(false); +#[cfg(unix)] +const OUTPUT_MODE: u32 = 0o600; + +pub(super) struct TelemetryLease; + +impl Drop for TelemetryLease { + fn drop(&mut self) { + TELEMETRY_LEASED.store(false, Ordering::Release); + } +} + +pub(super) fn acquire_telemetry_lease() -> Result { + TELEMETRY_LEASED + .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed) + .map(|_| TelemetryLease) + .map_err(|_| TelemetryProducerError::Busy) +} + +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum TelemetryOperation { + GetObject, + PutObject, + HeadObject, + ListObjects, + InternalRpc, +} + +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum TelemetrySpanStatus { + Ok, + Error, +} + +/// A classified observation supplied by a server-side adapter. +/// +/// Its closed shape deliberately has nowhere to retain a URL, header, object +/// name, trace attribute, or message body. The adapter must classify an event +/// before crossing this boundary. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ObservedTelemetrySpan { + operation: TelemetryOperation, + duration: Duration, + status: TelemetrySpanStatus, +} + +impl ObservedTelemetrySpan { + pub fn new(operation: TelemetryOperation, duration: Duration, status: TelemetrySpanStatus) -> Self { + Self { + operation, + duration, + status, + } + } +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub struct TelemetrySpan { + pub operation: TelemetryOperation, + pub duration_micros: u64, + pub status: TelemetrySpanStatus, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub struct RecordedTrace { + pub spans: Vec, + pub dropped_span_count: u64, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum TraceRecordCompletion { + Complete, + LimitExceeded, + SourceUnavailable, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct TraceRecordCapture { + pub data: RecordedTrace, + pub completion: TraceRecordCompletion, +} + +#[derive(Clone, Copy, Debug)] +pub struct LocalTelemetryConsent { + expires_at: Instant, +} + +impl LocalTelemetryConsent { + pub fn new(expires_at: Instant) -> Result { + if expires_at <= Instant::now() { + return Err(TelemetryProducerError::ConsentExpired); + } + Ok(Self { expires_at }) + } + + pub fn remaining(self) -> Result { + self.expires_at + .checked_duration_since(Instant::now()) + .filter(|remaining| !remaining.is_zero()) + .ok_or(TelemetryProducerError::ConsentExpired) + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct TelemetryArtifactConsent { + pub consent_uid: String, + pub policy_revision: u64, + pub expires_at_unix: i64, + pub confirmed: bool, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct TelemetryProvenance { + repository: &'static str, + source_commit: String, + executable_sha256: String, + rustfs_version: String, + os_family: TelemetryOsFamily, + architecture: TelemetryArchitecture, + build_features: Vec, +} + +impl TelemetryProvenance { + 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: TelemetryOsFamily::current(), + architecture: TelemetryArchitecture::current(), + build_features, + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +enum TelemetryOsFamily { + Linux, + Darwin, + Windows, + Freebsd, + Other, +} + +impl TelemetryOsFamily { + 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, Eq, PartialEq, Serialize)] +#[serde(rename_all = "lowercase")] +enum TelemetryArchitecture { + #[serde(rename = "x86_64")] + X86_64, + Aarch64, + Other, +} + +impl TelemetryArchitecture { + fn current() -> Self { + match std::env::consts::ARCH { + "x86_64" => Self::X86_64, + "aarch64" => Self::Aarch64, + _ => Self::Other, + } + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct TelemetryArtifactRequest { + 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 consent: TelemetryArtifactConsent, + pub produced_at_unix: i64, + pub expires_at_unix: i64, + pub nonce: [u8; 32], + pub provenance: TelemetryProvenance, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +pub enum TelemetryTool { + #[serde(rename = "telemetry.record")] + Record, + #[serde(rename = "telemetry.otlp")] + Otlp, + #[serde(rename = "telemetry.replay")] + Replay, +} + +impl TelemetryTool { + pub const fn id(self) -> &'static str { + match self { + Self::Record => "telemetry.record", + Self::Otlp => "telemetry.otlp", + Self::Replay => "telemetry.replay", + } + } + + pub const fn capability(self) -> &'static str { + match self { + Self::Record => TELEMETRY_RECORD_CAPABILITY, + Self::Otlp => TELEMETRY_OTLP_CAPABILITY, + Self::Replay => TELEMETRY_REPLAY_CAPABILITY, + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum TelemetryOutcome { + Succeeded, + Partial, + Failed, + Unsupported, + Cancelled, +} + +impl TelemetryOutcome { + 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, Eq, PartialEq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum TelemetryReasonCode { + Complete, + LimitExceeded, + SourceUnavailable, + PermissionDenied, + UnsupportedTool, + Cancelled, + InvalidInput, + CollectionFailed, + CounterReset, +} + +impl TelemetryReasonCode { + pub const fn as_str(self) -> &'static str { + match self { + Self::Complete => "COMPLETE", + Self::LimitExceeded => "LIMIT_EXCEEDED", + Self::SourceUnavailable => "SOURCE_UNAVAILABLE", + Self::PermissionDenied => "PERMISSION_DENIED", + Self::UnsupportedTool => "UNSUPPORTED_TOOL", + Self::Cancelled => "CANCELLED", + Self::InvalidInput => "INVALID_INPUT", + Self::CollectionFailed => "COLLECTION_FAILED", + Self::CounterReset => "COUNTER_RESET", + } + } +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct TelemetryCoverage { + requested_units: u64, + completed_units: u64, + unit: &'static str, +} + +impl TelemetryCoverage { + pub const fn window(completed: bool) -> Self { + Self { + requested_units: 1, + completed_units: completed as u64, + unit: "WINDOW", + } + } + + const fn partial_windows() -> Self { + Self { + requested_units: 2, + completed_units: 1, + unit: "WINDOW", + } + } +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct TelemetryDiagnosticResult { + schema_version: u16, + run_uid: String, + tool_id: TelemetryTool, + capability: &'static str, + outcome: TelemetryOutcome, + reason_code: TelemetryReasonCode, + duration_millis: u64, + provenance: TelemetryProvenance, + coverage: TelemetryCoverage, + data: Option, +} + +impl TelemetryDiagnosticResult { + pub fn succeeded(request: &TelemetryArtifactRequest, tool: TelemetryTool, duration: Duration, data: T) -> Self { + Self { + schema_version: TELEMETRY_SCHEMA_VERSION, + run_uid: request.run_uid.clone(), + tool_id: tool, + capability: tool.capability(), + outcome: TelemetryOutcome::Succeeded, + reason_code: TelemetryReasonCode::Complete, + duration_millis: bounded_duration_millis(duration), + provenance: request.provenance.clone(), + coverage: TelemetryCoverage::window(true), + data: Some(data), + } + } + + pub fn partial( + request: &TelemetryArtifactRequest, + tool: TelemetryTool, + duration: Duration, + reason_code: TelemetryReasonCode, + data: T, + ) -> Self { + Self { + schema_version: TELEMETRY_SCHEMA_VERSION, + run_uid: request.run_uid.clone(), + tool_id: tool, + capability: tool.capability(), + outcome: TelemetryOutcome::Partial, + reason_code, + duration_millis: bounded_duration_millis(duration), + provenance: request.provenance.clone(), + coverage: TelemetryCoverage::partial_windows(), + data: Some(data), + } + } + + fn failed( + request: &TelemetryArtifactRequest, + tool: TelemetryTool, + duration: Duration, + reason_code: TelemetryReasonCode, + ) -> Self { + Self { + schema_version: TELEMETRY_SCHEMA_VERSION, + run_uid: request.run_uid.clone(), + tool_id: tool, + capability: tool.capability(), + outcome: TelemetryOutcome::Failed, + reason_code, + duration_millis: bounded_duration_millis(duration), + provenance: request.provenance.clone(), + coverage: TelemetryCoverage::window(false), + data: None, + } + } + + pub fn unsupported(request: &TelemetryArtifactRequest, tool: TelemetryTool, duration: Duration) -> Self { + Self { + schema_version: TELEMETRY_SCHEMA_VERSION, + run_uid: request.run_uid.clone(), + tool_id: tool, + capability: tool.capability(), + outcome: TelemetryOutcome::Unsupported, + reason_code: TelemetryReasonCode::UnsupportedTool, + duration_millis: bounded_duration_millis(duration), + provenance: request.provenance.clone(), + coverage: TelemetryCoverage::window(false), + data: None, + } + } + + pub fn outcome(&self) -> TelemetryOutcome { + self.outcome + } + + pub fn reason_code(&self) -> TelemetryReasonCode { + self.reason_code + } + + pub fn data(&self) -> Option<&T> { + self.data.as_ref() + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct SignedTelemetryExport { + pub artifact_uid: String, + pub tool: TelemetryTool, + pub outcome: TelemetryOutcome, + pub reason_code: TelemetryReasonCode, + pub archive_bytes: Vec, + pub archive_sha256: String, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct SavedTelemetryExport { + pub artifact_uid: String, + pub archive_size_bytes: u64, + pub archive_sha256: String, +} + +#[derive(Clone, Copy, Debug)] +pub struct TraceRecordLimits { + pub duration: Duration, + pub max_spans: usize, +} + +impl TraceRecordLimits { + pub fn validate(self) -> Result { + if self.duration.is_zero() || self.duration > MAX_TELEMETRY_DURATION { + return Err(TelemetryProducerError::InvalidDuration); + } + if self.max_spans == 0 || self.max_spans > MAX_TELEMETRY_SPANS { + return Err(TelemetryProducerError::InvalidSpanLimit); + } + Ok(self) + } +} + +#[derive(Debug, Error, Eq, PartialEq)] +pub enum TelemetryProducerError { + #[error("another telemetry operation is already running")] + Busy, + #[error("local telemetry consent is expired")] + ConsentExpired, + #[error("telemetry duration must be between 1ns and 30s")] + InvalidDuration, + #[error("telemetry span limit must be between 1 and 1024")] + InvalidSpanLimit, + #[error("telemetry capture was cancelled")] + Cancelled, + #[error("telemetry source closed before the requested window completed")] + SourceUnavailable, + #[error("telemetry duration exceeds the contract integer range")] + DurationOverflow, + #[error("telemetry result exceeds 262144 bytes")] + ResultTooLarge, +} + +#[derive(Debug, Error)] +pub enum TelemetryArtifactError { + #[error("telemetry_invalid_request")] + InvalidRequest, + #[error("telemetry_unsupported_version")] + UnsupportedVersion, + #[error("telemetry_local_consent_required")] + ConsentRequired, + #[error("telemetry_local_consent_expired")] + ConsentExpired, + #[error("telemetry_request_expired")] + Expired, + #[error("telemetry_collection_cancelled")] + Cancelled, + #[error("telemetry_limit_exceeded")] + LimitExceeded, + #[error("telemetry_export_signing_failed")] + Signing, + #[error("telemetry_export_exists")] + AlreadyExists, + #[error("telemetry_export_encoding_failed")] + Encoding, + #[error("telemetry_export_io_failed")] + Io(#[source] std::io::Error), + #[error("telemetry_export_durability_failed_after_commit")] + DurabilityAfterCommit(#[source] std::io::Error), +} + +/// Capture classified spans until the bounded window ends. +/// +/// Reaching `max_spans` stops capture immediately and increments the drop count +/// for every already queued observation that could be counted without waiting. +pub async fn record_trace( + mut source: mpsc::Receiver, + consent: LocalTelemetryConsent, + limits: TraceRecordLimits, + cancel: &CancellationToken, +) -> Result { + let limits = limits.validate()?; + if cancel.is_cancelled() { + return Err(TelemetryProducerError::Cancelled); + } + let remaining = consent.remaining()?; + if remaining < limits.duration { + return Err(TelemetryProducerError::ConsentExpired); + } + let _lease = acquire_telemetry_lease()?; + let deadline = Instant::now() + limits.duration; + let mut spans = Vec::with_capacity(limits.max_spans.min(64)); + let mut dropped_span_count = 0u64; + let mut completion = TraceRecordCompletion::Complete; + + loop { + tokio::select! { + biased; + _ = cancel.cancelled() => return Err(TelemetryProducerError::Cancelled), + _ = tokio::time::sleep_until(deadline.into()) => break, + observed = source.recv() => { + let Some(observed) = observed else { + if spans.is_empty() { + return Err(TelemetryProducerError::SourceUnavailable); + } + completion = TraceRecordCompletion::SourceUnavailable; + break; + }; + if spans.len() == limits.max_spans { + completion = TraceRecordCompletion::LimitExceeded; + dropped_span_count = dropped_span_count.saturating_add(1).min(MAX_SAFE_INTEGER); + while source.try_recv().is_ok() { + dropped_span_count = dropped_span_count.saturating_add(1).min(MAX_SAFE_INTEGER); + } + break; + } + spans.push(to_contract_span(observed)?); + } + } + } + + let result = RecordedTrace { + spans, + dropped_span_count, + }; + ensure_result_size(&result)?; + Ok(TraceRecordCapture { + data: result, + completion, + }) +} + +/// Capture the process-local RustFS trace bus. +/// +/// The current bus exposes only heal and scanner operations, none of which has +/// the frozen GET/PUT/HEAD/LIST/INTERNAL_RPC semantics. The adapter consumes +/// that real source but refuses to infer an operation or status from raw +/// fields, so it remains unavailable until RustFS publishes an approved typed +/// event. +pub async fn record_trace_bus( + consent: LocalTelemetryConsent, + limits: TraceRecordLimits, + cancel: &CancellationToken, +) -> Result { + let limits = limits.validate()?; + if cancel.is_cancelled() { + return Err(TelemetryProducerError::Cancelled); + } + let remaining = consent.remaining()?; + if remaining < limits.duration { + return Err(TelemetryProducerError::ConsentExpired); + } + let _lease = acquire_telemetry_lease()?; + let deadline = Instant::now() + limits.duration; + let mut subscription = subscribe_trace_events(); + loop { + tokio::select! { + biased; + _ = cancel.cancelled() => return Err(TelemetryProducerError::Cancelled), + _ = tokio::time::sleep_until(deadline.into()) => return Err(TelemetryProducerError::SourceUnavailable), + received = subscription.recv() => match received { + Ok(_event) => {} + Err(tokio::sync::broadcast::error::RecvError::Lagged(_dropped)) => {} + Err(tokio::sync::broadcast::error::RecvError::Closed) => return Err(TelemetryProducerError::SourceUnavailable), + } + } + } +} + +pub fn record_diagnostic_result( + request: &TelemetryArtifactRequest, + capture: TraceRecordCapture, + duration: Duration, +) -> TelemetryDiagnosticResult { + match capture.completion { + TraceRecordCompletion::Complete => { + TelemetryDiagnosticResult::succeeded(request, TelemetryTool::Record, duration, capture.data) + } + TraceRecordCompletion::LimitExceeded => { + TelemetryDiagnosticResult::failed(request, TelemetryTool::Record, duration, TelemetryReasonCode::CollectionFailed) + } + TraceRecordCompletion::SourceUnavailable => { + TelemetryDiagnosticResult::failed(request, TelemetryTool::Record, duration, TelemetryReasonCode::SourceUnavailable) + } + } +} + +fn to_contract_span(observed: ObservedTelemetrySpan) -> Result { + let duration_micros = u64::try_from(observed.duration.as_micros()).map_err(|_| TelemetryProducerError::DurationOverflow)?; + if duration_micros > MAX_SAFE_INTEGER { + return Err(TelemetryProducerError::DurationOverflow); + } + Ok(TelemetrySpan { + operation: observed.operation, + duration_micros, + status: observed.status, + }) +} + +pub(crate) fn ensure_result_size(value: &impl Serialize) -> Result<(), TelemetryProducerError> { + let bytes = serde_json::to_vec(value).map_err(|_| TelemetryProducerError::ResultTooLarge)?; + if bytes.len() > MAX_TELEMETRY_RESULT_BYTES { + return Err(TelemetryProducerError::ResultTooLarge); + } + Ok(()) +} + +pub fn encode_signed_telemetry_export( + request: &TelemetryArtifactRequest, + result: &TelemetryDiagnosticResult, + key: &DeviceIdentity, + cancel: &CancellationToken, +) -> Result { + request.validate()?; + check_cancel(cancel)?; + if result.schema_version != request.schema_version || result.run_uid != request.run_uid { + return Err(TelemetryArtifactError::InvalidRequest); + } + if !matches!(result.outcome, TelemetryOutcome::Succeeded | TelemetryOutcome::Partial) { + return Err(TelemetryArtifactError::InvalidRequest); + } + let valid_publishable_shape = match result.outcome { + TelemetryOutcome::Succeeded => { + result.data.is_some() + && result.reason_code == TelemetryReasonCode::Complete + && result.coverage.requested_units == 1 + && result.coverage.completed_units == 1 + && result.coverage.unit == "WINDOW" + } + TelemetryOutcome::Partial => { + result.data.is_some() + && matches!( + result.reason_code, + TelemetryReasonCode::LimitExceeded + | TelemetryReasonCode::SourceUnavailable + | TelemetryReasonCode::CounterReset + ) + && result.coverage.requested_units == 2 + && result.coverage.completed_units == 1 + && result.coverage.unit == "WINDOW" + } + _ => false, + }; + if !valid_publishable_shape { + return Err(TelemetryArtifactError::InvalidRequest); + } + if result.capability != result.tool_id.capability() { + return Err(TelemetryArtifactError::InvalidRequest); + } + let result_bytes = serde_json::to_vec(result).map_err(|_| TelemetryArtifactError::Encoding)?; + if result_bytes.is_empty() || result_bytes.len() > MAX_TELEMETRY_RESULT_BYTES { + return Err(TelemetryArtifactError::LimitExceeded); + } + let device_key_id = hex_lower(&Sha256::digest(key.public_key_der())); + let envelope = TelemetryEnvelope { + 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: TELEMETRY_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: TelemetryPayload { + 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(|_| TelemetryArtifactError::Encoding)?; + if envelope_bytes.is_empty() || envelope_bytes.len() > MAX_ENVELOPE_BYTES { + return Err(TelemetryArtifactError::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(TelemetryArtifactError::LimitExceeded)?; + if decompressed > MAX_DECOMPRESSED_BYTES { + return Err(TelemetryArtifactError::LimitExceeded); + } + check_cancel(cancel)?; + if unix_now()? >= request.expires_at_unix { + return Err(TelemetryArtifactError::Expired); + } + let archive_bytes = archive(&envelope_bytes, &signature_bytes, &result_bytes)?; + if archive_bytes.len() > MAX_ARCHIVE_BYTES { + return Err(TelemetryArtifactError::LimitExceeded); + } + Ok(SignedTelemetryExport { + artifact_uid: request.artifact_uid.clone(), + tool: result.tool_id, + outcome: result.outcome, + reason_code: result.reason_code, + archive_sha256: hex_lower(&Sha256::digest(&archive_bytes)), + archive_bytes, + }) +} + +pub fn save_signed_telemetry_export( + output: &Path, + export: &SignedTelemetryExport, + cancel: &CancellationToken, +) -> Result { + check_cancel(cancel)?; + if !uuid7(&export.artifact_uid) + || export.archive_bytes.is_empty() + || export.archive_bytes.len() > MAX_ARCHIVE_BYTES + || hex_lower(&Sha256::digest(&export.archive_bytes)) != export.archive_sha256 + { + return Err(TelemetryArtifactError::InvalidRequest); + } + let parent = output + .parent() + .filter(|path| !path.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + let filename = output + .file_name() + .ok_or(TelemetryArtifactError::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(TelemetryArtifactError::Io)?; + check_cancel(cancel)?; + file.sync_all().map_err(TelemetryArtifactError::Io)?; + check_cancel(cancel)?; + fs::hard_link(&temporary, output).map_err(map_publish_error)?; + if let Err(error) = fs::remove_file(&temporary) { + return Err(TelemetryArtifactError::DurabilityAfterCommit(error)); + } + #[cfg(unix)] + if let Err(error) = File::open(parent).and_then(|directory| directory.sync_all()) { + return Err(TelemetryArtifactError::DurabilityAfterCommit(error)); + } + Ok(SavedTelemetryExport { + 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 TelemetryArtifactRequest { + pub fn validate(&self) -> Result<(), TelemetryArtifactError> { + self.validate_at(unix_now()?) + } + + fn validate_at(&self, now_unix: i64) -> Result<(), TelemetryArtifactError> { + if self.schema_version != TELEMETRY_SCHEMA_VERSION { + return Err(TelemetryArtifactError::UnsupportedVersion); + } + if !self.consent.confirmed || self.consent.policy_revision == 0 { + return Err(TelemetryArtifactError::ConsentRequired); + } + if self.consent.expires_at_unix <= now_unix || self.expires_at_unix > self.consent.expires_at_unix { + return Err(TelemetryArtifactError::ConsentExpired); + } + let validity = self + .expires_at_unix + .checked_sub(self.produced_at_unix) + .ok_or(TelemetryArtifactError::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(TelemetryArtifactError::Expired); + } + 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(TelemetryArtifactError::InvalidRequest); + } + Ok(()) + } +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct TelemetryEnvelope<'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: TelemetryTool, + 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: TelemetryPayload, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct TelemetryPayload { + path: &'static str, + media_type: &'static str, + size_bytes: u64, + sha256: String, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct SignatureDocument<'a> { + algorithm: &'static str, + key_id: &'a str, + value: String, +} + +fn signature_document(key: &DeviceIdentity, key_id: &str, envelope: &[u8]) -> Result, TelemetryArtifactError> { + let pkcs8 = key.to_pkcs8_der().map_err(|_| TelemetryArtifactError::Signing)?; + let signing_key = SigningKey::from_pkcs8_der(pkcs8.as_slice()).map_err(|_| TelemetryArtifactError::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(&SignatureDocument { + algorithm: "ES256", + key_id, + value: URL_SAFE_NO_PAD.encode_to_string(signature.normalize_s().to_bytes()), + }) + .map_err(|_| TelemetryArtifactError::Encoding) +} + +fn archive(envelope: &[u8], signature: &[u8], result: &[u8]) -> Result, TelemetryArtifactError> { + 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(|_| TelemetryArtifactError::Encoding)?; + writer.write_all(bytes).map_err(TelemetryArtifactError::Io)?; + } + writer + .finish() + .map(|cursor| cursor.into_inner()) + .map_err(|_| TelemetryArtifactError::Encoding) +} + +fn resource_names_match(request: &TelemetryArtifactRequest) -> 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 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 build_feature(value: &str) -> bool { + !value.is_empty() + && value.len() <= 64 + && value.as_bytes()[0].is_ascii_lowercase() + && value + .bytes() + .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'_' | b'-')) +} + +fn timestamp(unix: i64) -> Result { + OffsetDateTime::from_unix_timestamp(unix) + .map_err(|_| TelemetryArtifactError::InvalidRequest)? + .format(&Rfc3339) + .map_err(|_| TelemetryArtifactError::InvalidRequest) +} + +fn unix_now() -> Result { + let duration = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|_| TelemetryArtifactError::InvalidRequest)?; + i64::try_from(duration.as_secs()).map_err(|_| TelemetryArtifactError::InvalidRequest) +} + +fn check_cancel(cancel: &CancellationToken) -> Result<(), TelemetryArtifactError> { + if cancel.is_cancelled() { + Err(TelemetryArtifactError::Cancelled) + } else { + Ok(()) + } +} + +fn bounded_duration_millis(duration: Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX).min(30_000) +} + +fn hex_lower(bytes: &[u8]) -> String { + hex_simd::encode_to_string(bytes, hex_simd::AsciiCase::Lower) +} + +fn map_create_error(error: std::io::Error) -> TelemetryArtifactError { + if error.kind() == std::io::ErrorKind::AlreadyExists { + TelemetryArtifactError::AlreadyExists + } else { + TelemetryArtifactError::Io(error) + } +} + +fn map_publish_error(error: std::io::Error) -> TelemetryArtifactError { + if error.kind() == std::io::ErrorKind::AlreadyExists { + TelemetryArtifactError::AlreadyExists + } else { + TelemetryArtifactError::Io(error) + } +} diff --git a/rustfs/src/connect/diagnostics/trace_replay.rs b/rustfs/src/connect/diagnostics/trace_replay.rs new file mode 100644 index 000000000..9690ac4d8 --- /dev/null +++ b/rustfs/src/connect/diagnostics/trace_replay.rs @@ -0,0 +1,127 @@ +// 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. + +//! In-memory replay of a locally reviewed telemetry record. +//! +//! This module never opens a path and never replays S3 operations. The caller +//! supplies reviewed bytes; replay validates the exact closed record shape and +//! returns only its redacted operation/timing/status observations. + +use std::time::Instant; + +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use thiserror::Error; +use tokio_util::sync::CancellationToken; + +use super::trace_record::{ + LocalTelemetryConsent, MAX_SAFE_INTEGER, MAX_TELEMETRY_RESULT_BYTES, MAX_TELEMETRY_SPANS, RecordedTrace, + TelemetryArtifactRequest, TelemetryDiagnosticResult, TelemetryProducerError, TelemetrySpan, TelemetryTool, + acquire_telemetry_lease, ensure_result_size, +}; + +pub struct LocallyReviewedTraceArtifact<'a> { + bytes: &'a [u8], +} + +impl<'a> LocallyReviewedTraceArtifact<'a> { + pub fn new(bytes: &'a [u8]) -> Result { + if bytes.is_empty() || bytes.len() > MAX_TELEMETRY_RESULT_BYTES { + return Err(TraceReplayError::InvalidArtifact); + } + Ok(Self { bytes }) + } +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub struct ReplayedTrace { + pub input_artifact_sha256: String, + pub spans: Vec, +} + +#[derive(Debug, Error, Eq, PartialEq)] +pub enum TraceReplayError { + #[error("another telemetry operation is already running")] + Busy, + #[error("reviewed telemetry artifact is invalid")] + InvalidArtifact, + #[error("telemetry replay was cancelled")] + Cancelled, + #[error("local telemetry consent is expired")] + ConsentExpired, + #[error("telemetry replay result exceeds 262144 bytes")] + ResultTooLarge, +} + +pub fn replay_trace( + artifact: LocallyReviewedTraceArtifact<'_>, + consent: LocalTelemetryConsent, + cancel: &CancellationToken, +) -> Result { + consent.remaining().map_err(map_consent_error)?; + if cancel.is_cancelled() { + return Err(TraceReplayError::Cancelled); + } + let _lease = acquire_telemetry_lease().map_err(map_consent_error)?; + let record: RecordedTrace = serde_json::from_slice(artifact.bytes).map_err(|_| TraceReplayError::InvalidArtifact)?; + if record.spans.len() > MAX_TELEMETRY_SPANS + || record.dropped_span_count > MAX_SAFE_INTEGER + || record.spans.iter().any(|span| span.duration_micros > MAX_SAFE_INTEGER) + { + return Err(TraceReplayError::InvalidArtifact); + } + if cancel.is_cancelled() { + return Err(TraceReplayError::Cancelled); + } + let result = ReplayedTrace { + input_artifact_sha256: hex_lower(&Sha256::digest(artifact.bytes)), + spans: record.spans, + }; + ensure_result_size(&result).map_err(|_| TraceReplayError::ResultTooLarge)?; + Ok(result) +} + +pub fn replay_trace_result( + request: &TelemetryArtifactRequest, + artifact: LocallyReviewedTraceArtifact<'_>, + consent: LocalTelemetryConsent, + cancel: &CancellationToken, +) -> Result, TraceReplayError> { + let started = Instant::now(); + let replay = replay_trace(artifact, consent, cancel)?; + Ok(TelemetryDiagnosticResult::succeeded( + request, + TelemetryTool::Replay, + started.elapsed(), + replay, + )) +} + +fn map_consent_error(error: TelemetryProducerError) -> TraceReplayError { + match error { + TelemetryProducerError::Busy => TraceReplayError::Busy, + TelemetryProducerError::ConsentExpired => TraceReplayError::ConsentExpired, + _ => TraceReplayError::InvalidArtifact, + } +} + +fn hex_lower(bytes: &[u8]) -> String { + use std::fmt::Write as _; + + bytes.iter().fold(String::with_capacity(bytes.len() * 2), |mut output, byte| { + let _ = write!(output, "{byte:02x}"); + output + }) +} diff --git a/rustfs/src/connect/mod.rs b/rustfs/src/connect/mod.rs index 1991d2b13..91c06a115 100644 --- a/rustfs/src/connect/mod.rs +++ b/rustfs/src/connect/mod.rs @@ -46,13 +46,22 @@ pub use config::{HeartbeatConfig, HeartbeatConfigError, HeartbeatSchedule}; pub use credential_store::{CredentialStore, DeviceCredential}; pub use diagnostics::{ CPU_PROFILE_CAPABILITY, CaptureMode, DiagnosticCollectionPolicy, DiagnosticReceipt, DiagnosticScheduleError, - DiagnosticScheduleRuntime, DiagnosticScheduleStatus, LOGS_CAPABILITY, LOGS_SCHEMA_VERSION, LocalLogConsent, - LocalProfileConsent, LogCaptureError, LogCaptureRequest, LogProvenance, MAX_PROFILE_DURATION, MEMORY_PROFILE_CAPABILITY, - PROFILE_SCHEMA_VERSION, ProfileCaptureRequest, ProfileData, ProfileError, ProfileOutcome, ProfileProvenance, - ProfileReasonCode, ProfileResult, ProfileTool, ReceiptOutcome, SavedLogExport, SavedProfileExport, SignedLogExport, - SignedProfileExport, THREAD_PROFILE_CAPABILITY, ThreadProfileScope, capture_cpu_profile, capture_thread_profile, - encode_signed_profile_export, export_cpu_profile, export_logs, export_memory_profile, export_thread_profile, - run_local_environment_once, save_signed_log_export, save_signed_profile_export, spawn_environment_schedule, + DiagnosticScheduleRuntime, DiagnosticScheduleStatus, LOGS_CAPABILITY, LOGS_SCHEMA_VERSION, LocalLogConsent, LocalOtlpHeaders, + LocalProfileConsent, LocalTelemetryConsent, LocallyReviewedTraceArtifact, LogCaptureError, LogCaptureRequest, LogProvenance, + MAX_OTLP_BODY_BYTES, MAX_PROFILE_DURATION, MAX_SAFE_INTEGER, MAX_TELEMETRY_DURATION, MAX_TELEMETRY_RESULT_BYTES, + MAX_TELEMETRY_SPANS, MEMORY_PROFILE_CAPABILITY, ObservedTelemetrySpan, OperationSummary, OtlpBatch, OtlpForwardError, + OtlpReceipt, PROFILE_SCHEMA_VERSION, ProfileCaptureRequest, ProfileData, ProfileError, ProfileOutcome, ProfileProvenance, + ProfileReasonCode, ProfileResult, ProfileTool, ReceiptOutcome, RecordedTrace, ReplayedTrace, SavedLogExport, + SavedProfileExport, SavedTelemetryExport, SignedLogExport, SignedProfileExport, SignedTelemetryExport, + TELEMETRY_OTLP_CAPABILITY, TELEMETRY_RECORD_CAPABILITY, TELEMETRY_REPLAY_CAPABILITY, TELEMETRY_SCHEMA_VERSION, + THREAD_PROFILE_CAPABILITY, TelemetryArtifactConsent, TelemetryArtifactError, TelemetryArtifactRequest, TelemetryCoverage, + TelemetryDiagnosticResult, TelemetryOperation, TelemetryOutcome, TelemetryProducerError, TelemetryProvenance, + TelemetryReasonCode, TelemetrySpan, TelemetrySpanStatus, TelemetryTool, ThreadProfileScope, TraceAnalysis, + TraceAnalysisError, TraceRecordCapture, TraceRecordCompletion, TraceRecordLimits, TraceReplayError, analyze_trace, + capture_cpu_profile, capture_thread_profile, encode_signed_profile_export, encode_signed_telemetry_export, + export_cpu_profile, export_logs, export_memory_profile, export_thread_profile, export_trace_otlp, export_trace_otlp_result, + record_diagnostic_result, record_trace, record_trace_bus, replay_trace, replay_trace_result, run_local_environment_once, + save_signed_log_export, save_signed_profile_export, save_signed_telemetry_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 83e1286c2..2fdf83f54 100644 --- a/rustfs/src/startup_entrypoint.rs +++ b/rustfs/src/startup_entrypoint.rs @@ -15,7 +15,8 @@ use crate::{ config::{ CommandResult, Config, ConnectLicenseCommands, ConnectLicenseScopeOpts, ConnectLogsMode, ConnectLogsOpts, - ConnectProfileOpts, ConnectProfileTool, ConnectThreadProfileScope, Opt, + ConnectProfileOpts, ConnectProfileTool, ConnectTelemetryArtifactOpts, ConnectTelemetryCommands, + ConnectThreadProfileScope, Opt, }, startup_lifecycle::{StartupRuntimeLifecycle, run_startup_runtime_lifecycle}, startup_preflight::{StartupServerPreflightError, bootstrap_external_prefix_compat, init_startup_server_preflight}, @@ -26,7 +27,8 @@ use crate::{ storage_api::startup::storage::bootstrap_instance_ctx, }; use std::io::{Error, Read as _, Result}; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; +use tokio_util::sync::CancellationToken; use tracing::{error, instrument}; const LOG_COMPONENT_MAIN: &str = "main"; @@ -136,6 +138,7 @@ async fn async_main() -> Result<()> { CommandResult::ConnectLicense(command) => return execute_connect_license(command), CommandResult::ConnectProfile(options) => return execute_connect_profile(options).await, CommandResult::ConnectLogs(options) => return execute_connect_logs(options).await, + CommandResult::ConnectTelemetry(command) => return execute_connect_telemetry(command).await, CommandResult::Server(config) => config, }; @@ -247,6 +250,231 @@ async fn execute_connect_logs(options: ConnectLogsOpts) -> Result<()> { Ok(()) } +async fn execute_connect_telemetry(command: ConnectTelemetryCommands) -> Result<()> { + use crate::connect::{ + LocalOtlpHeaders, LocallyReviewedTraceArtifact, MAX_OTLP_BODY_BYTES, MAX_TELEMETRY_RESULT_BYTES, OtlpBatch, + RecordedTrace, TelemetryDiagnosticResult, TelemetryProducerError, TelemetryTool, TraceRecordLimits, analyze_trace, + export_trace_otlp_result, record_diagnostic_result, record_trace_bus, replay_trace_result, + }; + use reqwest::header::{AUTHORIZATION, HeaderMap, HeaderValue}; + + match command { + ConnectTelemetryCommands::Record(options) => { + let (key, request, consent) = telemetry_context(&options.artifact)?; + request.validate().map_err(Error::other)?; + if options.duration_millis == 0 + || options.duration_millis > 30_000 + || options.max_spans == 0 + || options.max_spans > 1_024 + { + return Err(Error::other("telemetry record limits are invalid")); + } + let cancel = CancellationToken::new(); + let started = Instant::now(); + let capture = record_trace_bus( + consent, + TraceRecordLimits { + duration: Duration::from_millis(options.duration_millis), + max_spans: options.max_spans, + }, + &cancel, + ); + tokio::pin!(capture); + let capture = tokio::select! { + biased; + signal = tokio::signal::ctrl_c() => { + signal.map_err(Error::other)?; + cancel.cancel(); + return Err(Error::other("telemetry record cancelled")); + } + result = capture.as_mut() => result, + }; + match capture { + Ok(capture) => { + let result = record_diagnostic_result(&request, capture, started.elapsed()); + save_telemetry_result(&options.artifact, &request, &result, &key, &cancel, None) + } + Err(TelemetryProducerError::SourceUnavailable) => { + let result = TelemetryDiagnosticResult::::unsupported( + &request, + TelemetryTool::Record, + started.elapsed(), + ); + println!("{}", serde_json::to_string(&result).map_err(Error::other)?); + Ok(()) + } + Err(error) => Err(Error::other(error)), + } + } + ConnectTelemetryCommands::Otlp(options) => { + let (key, request, consent) = telemetry_context(&options.artifact)?; + request.validate().map_err(Error::other)?; + let body = read_bounded_stdin(MAX_OTLP_BODY_BYTES)?; + let batch = OtlpBatch::new(body).map_err(Error::other)?; + let endpoint = reqwest::Url::parse(&options.endpoint).map_err(|_| Error::other("invalid OTLP endpoint"))?; + let mut headers = HeaderMap::new(); + if let Some(name) = options.authorization_env.as_deref() { + if !valid_environment_name(name) { + return Err(Error::other("invalid OTLP authorization environment variable name")); + } + let value = std::env::var(name).map_err(|_| Error::other("OTLP authorization is unavailable"))?; + let value = HeaderValue::from_str(&value).map_err(|_| Error::other("OTLP authorization is invalid"))?; + headers.insert(AUTHORIZATION, value); + } + let cancel = CancellationToken::new(); + let result = export_trace_otlp_result( + &request, + endpoint, + LocalOtlpHeaders::new(headers), + batch, + consent, + Duration::from_millis(options.timeout_millis), + &cancel, + ) + .await + .map_err(Error::other)?; + save_telemetry_result(&options.artifact, &request, &result, &key, &cancel, None) + } + ConnectTelemetryCommands::Replay(options) => { + let (key, request, consent) = telemetry_context(&options.artifact)?; + request.validate().map_err(Error::other)?; + let bytes = read_bounded_stdin(MAX_TELEMETRY_RESULT_BYTES)?; + let cancel = CancellationToken::new(); + let result = replay_trace_result( + &request, + LocallyReviewedTraceArtifact::new(&bytes).map_err(Error::other)?, + consent, + &cancel, + ) + .map_err(Error::other)?; + let analysis = analyze_trace( + result + .data() + .ok_or_else(|| Error::other("telemetry replay returned no data"))?, + consent, + &cancel, + ) + .map_err(Error::other)?; + save_telemetry_result( + &options.artifact, + &request, + &result, + &key, + &cancel, + Some(serde_json::to_value(analysis).map_err(Error::other)?), + ) + } + } +} + +fn telemetry_context( + options: &ConnectTelemetryArtifactOpts, +) -> Result<( + crate::connect::DeviceIdentity, + crate::connect::TelemetryArtifactRequest, + crate::connect::LocalTelemetryConsent, +)> { + use crate::connect::{ + IdentityStore, LocalTelemetryConsent, TelemetryArtifactConsent, TelemetryArtifactRequest, TelemetryProvenance, + }; + 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 telemetry requires an enrolled device identity"))?; + let executable_sha256 = hash_current_executable()?; + let produced_at_unix = unix_now()?; + let remaining = options + .consent_expires_at_unix + .checked_sub(produced_at_unix) + .and_then(|seconds| u64::try_from(seconds).ok()) + .filter(|seconds| *seconds > 0) + .ok_or_else(|| Error::other("local telemetry consent is expired"))?; + let consent = LocalTelemetryConsent::new( + Instant::now() + .checked_add(Duration::from_secs(remaining)) + .ok_or_else(|| Error::other("local telemetry consent is expired"))?, + ) + .map_err(Error::other)?; + let mut nonce = [0_u8; 32]; + SysRng.try_fill_bytes(&mut nonce).map_err(Error::other)?; + let request = TelemetryArtifactRequest { + organization_name: options.organization.clone(), + cluster_name: options.cluster.clone(), + device_name: options.device.clone(), + run_uid: options.run_uid.clone(), + artifact_uid: options.artifact_uid.clone(), + schema_version: crate::connect::TELEMETRY_SCHEMA_VERSION, + consent: TelemetryArtifactConsent { + consent_uid: options.consent_uid.clone(), + 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, + provenance: TelemetryProvenance::new( + crate::version::build::COMMIT_HASH, + executable_sha256, + env!("CARGO_PKG_VERSION"), + enabled_build_features(), + ), + }; + Ok((key, request, consent)) +} + +fn save_telemetry_result( + options: &ConnectTelemetryArtifactOpts, + request: &crate::connect::TelemetryArtifactRequest, + result: &crate::connect::TelemetryDiagnosticResult, + key: &crate::connect::DeviceIdentity, + cancel: &CancellationToken, + analysis: Option, +) -> Result<()> { + let export = crate::connect::encode_signed_telemetry_export(request, result, key, cancel).map_err(Error::other)?; + let receipt = crate::connect::save_signed_telemetry_export(&options.output, &export, cancel).map_err(Error::other)?; + println!( + "{}", + serde_json::json!({ + "toolId": export.tool.id(), + "outcome": export.outcome.as_str(), + "reasonCode": export.reason_code.as_str(), + "artifactUid": receipt.artifact_uid, + "archiveSizeBytes": receipt.archive_size_bytes, + "archiveSha256": receipt.archive_sha256, + "analysis": analysis, + "upload": "NOT_PERFORMED", + }) + ); + Ok(()) +} + +fn read_bounded_stdin(limit: usize) -> Result> { + let mut bytes = Vec::with_capacity(limit.min(64 * 1024)); + std::io::stdin() + .take(u64::try_from(limit).unwrap_or(u64::MAX).saturating_add(1)) + .read_to_end(&mut bytes)?; + if bytes.is_empty() || bytes.len() > limit { + return Err(Error::other("telemetry stdin is empty or exceeds its limit")); + } + Ok(bytes) +} + +fn valid_environment_name(value: &str) -> bool { + !value.is_empty() + && value.len() <= 64 + && value + .bytes() + .all(|byte| byte.is_ascii_uppercase() || byte.is_ascii_digit() || byte == b'_') +} + +fn unix_now() -> Result { + let duration = SystemTime::now().duration_since(UNIX_EPOCH).map_err(Error::other)?; + i64::try_from(duration.as_secs()).map_err(Error::other) +} + async fn execute_connect_profile(options: ConnectProfileOpts) -> Result<()> { use crate::connect::{ IdentityStore, LocalProfileConsent, ProfileCaptureRequest, ProfileProvenance, ThreadProfileScope, export_cpu_profile, diff --git a/rustfs/tests/connect_trace_analysis.rs b/rustfs/tests/connect_trace_analysis.rs new file mode 100644 index 000000000..5e529236c --- /dev/null +++ b/rustfs/tests/connect_trace_analysis.rs @@ -0,0 +1,46 @@ +use serial_test::serial; +use std::time::{Duration, Instant}; + +use rustfs::connect::{LocalTelemetryConsent, LocallyReviewedTraceArtifact, TraceAnalysisError, analyze_trace, replay_trace}; +use tokio_util::sync::CancellationToken; + +fn consent() -> LocalTelemetryConsent { + LocalTelemetryConsent::new(Instant::now() + Duration::from_secs(1)).expect("future consent") +} + +#[test] +#[serial] +fn connect_trace_analysis_is_deterministic_and_uses_only_observed_values() { + let bytes = br#"{"spans":[{"operation":"GET_OBJECT","durationMicros":500,"status":"OK"},{"operation":"GET_OBJECT","durationMicros":250,"status":"ERROR"},{"operation":"INTERNAL_RPC","durationMicros":25,"status":"OK"}],"droppedSpanCount":0}"#; + let replay = replay_trace( + LocallyReviewedTraceArtifact::new(bytes).expect("bounded artifact"), + consent(), + &CancellationToken::new(), + ) + .expect("valid replay"); + let first = analyze_trace(&replay, consent(), &CancellationToken::new()).expect("analysis succeeds"); + let second = analyze_trace(&replay, consent(), &CancellationToken::new()).expect("analysis is repeatable"); + + assert_eq!(first, second); + assert_eq!(first.span_count, 3); + assert_eq!(first.error_count, 1); + assert_eq!(first.total_duration_micros, 775); + assert_eq!(first.operations.len(), 2); + assert_eq!(first.operations[0].span_count, 2); + assert_eq!(first.operations[0].error_count, 1); +} + +#[test] +#[serial] +fn connect_trace_analysis_honors_stop() { + let replay = replay_trace( + LocallyReviewedTraceArtifact::new(br#"{"spans":[],"droppedSpanCount":0}"#).expect("bounded artifact"), + consent(), + &CancellationToken::new(), + ) + .expect("valid replay"); + let cancel = CancellationToken::new(); + cancel.cancel(); + let error = analyze_trace(&replay, consent(), &cancel).expect_err("cancelled analysis must fail"); + assert_eq!(error, TraceAnalysisError::Cancelled); +} diff --git a/rustfs/tests/connect_trace_otlp.rs b/rustfs/tests/connect_trace_otlp.rs new file mode 100644 index 000000000..01cf6f4d5 --- /dev/null +++ b/rustfs/tests/connect_trace_otlp.rs @@ -0,0 +1,220 @@ +use serial_test::serial; +use std::time::{Duration, Instant}; + +use opentelemetry_proto::tonic::{ + collector::trace::v1::ExportTraceServiceRequest, + trace::v1::{ResourceSpans, ScopeSpans, Span}, +}; +use prost::Message as _; +use reqwest::{Url, header}; +use rustfs::connect::{ + LocalOtlpHeaders, LocalTelemetryConsent, LocallyReviewedTraceArtifact, OtlpBatch, OtlpForwardError, TraceReplayError, + export_trace_otlp, replay_trace, +}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpListener, +}; +use tokio_util::sync::CancellationToken; + +fn consent() -> LocalTelemetryConsent { + LocalTelemetryConsent::new(Instant::now() + Duration::from_secs(2)).expect("future consent") +} + +fn encoded_batch(span_count: usize) -> Vec { + ExportTraceServiceRequest { + resource_spans: vec![ResourceSpans { + scope_spans: vec![ScopeSpans { + spans: vec![Span::default(); span_count], + ..Default::default() + }], + ..Default::default() + }], + } + .encode_to_vec() +} + +fn encoded_named_batch(name_length: usize) -> Vec { + let mut span = Span::default(); + span.name = "x".repeat(name_length); + ExportTraceServiceRequest { + resource_spans: vec![ResourceSpans { + scope_spans: vec![ScopeSpans { + spans: vec![span], + ..Default::default() + }], + ..Default::default() + }], + } + .encode_to_vec() +} + +async fn collector(status: &str) -> (Url, tokio::task::JoinHandle>) { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind collector"); + let address = listener.local_addr().expect("collector address"); + let status = status.to_owned(); + let task = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("accept OTLP request"); + let mut received = vec![0u8; 4096]; + let size = stream.read(&mut received).await.expect("read OTLP request"); + received.truncate(size); + let response = format!("HTTP/1.1 {status}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"); + stream.write_all(response.as_bytes()).await.expect("write response"); + received + }); + (Url::parse(&format!("http://{address}/v1/traces")).expect("collector URL"), task) +} + +#[tokio::test] +#[serial] +async fn connect_trace_otlp_forwards_bounded_protobuf_and_keeps_auth_out_of_receipt() { + let (endpoint, server) = collector("200 OK").await; + let mut headers = header::HeaderMap::new(); + headers.insert(header::AUTHORIZATION, header::HeaderValue::from_static("Bearer local-secret")); + let receipt = export_trace_otlp( + endpoint, + LocalOtlpHeaders::new(headers), + OtlpBatch::new(encoded_batch(2)).expect("valid batch"), + consent(), + Duration::from_secs(1), + &CancellationToken::new(), + ) + .await + .expect("collector accepts batch"); + let request = String::from_utf8(server.await.expect("collector task")).expect("HTTP is UTF-8"); + assert!(request.to_ascii_lowercase().contains("authorization: bearer local-secret")); + assert!(request.to_ascii_lowercase().contains("content-type: application/x-protobuf")); + assert_eq!(receipt.accepted_span_count, 2); + assert_eq!(receipt.exported_bytes, encoded_batch(2).len() as u64); + assert!( + !serde_json::to_string(&receipt) + .expect("receipt JSON") + .contains("local-secret") + ); +} + +#[tokio::test] +#[serial] +async fn connect_trace_otlp_reports_fixed_failure_without_response_or_secret_body() { + let (endpoint, server) = collector("401 Unauthorized").await; + let error = export_trace_otlp( + endpoint, + LocalOtlpHeaders::new(header::HeaderMap::new()), + OtlpBatch::new(encoded_batch(1)).expect("valid batch"), + consent(), + Duration::from_secs(1), + &CancellationToken::new(), + ) + .await + .expect_err("rejection must fail"); + server.await.expect("collector task"); + assert_eq!(error, OtlpForwardError::Rejected); +} + +#[test] +#[serial] +fn connect_trace_otlp_rejects_remote_plaintext_and_oversized_batches() { + assert!(OtlpBatch::new(encoded_batch(1)).is_ok()); + assert!(OtlpBatch::new(encoded_batch(1024)).is_ok()); + assert!(OtlpBatch::new(encoded_batch(0)).is_err()); + assert!(OtlpBatch::new(encoded_batch(1025)).is_err()); + assert!(OtlpBatch::new(vec![0; 1_048_577]).is_err()); + assert!(OtlpBatch::new(vec![1]).is_err()); + let endpoint = Url::parse("http://collector.example/v1/traces").expect("URL"); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime"); + let error = runtime + .block_on(export_trace_otlp( + endpoint, + LocalOtlpHeaders::new(header::HeaderMap::new()), + OtlpBatch::new(encoded_batch(1)).expect("valid batch"), + consent(), + Duration::from_secs(1), + &CancellationToken::new(), + )) + .expect_err("plaintext remote collector must fail"); + assert_eq!(error, OtlpForwardError::Endpoint); +} + +#[test] +#[serial] +fn connect_trace_otlp_lease_blocks_replay_until_the_batch_is_dropped() { + let held = OtlpBatch::new(encoded_batch(1)).expect("held parsed batch"); + let reviewed = br#"{"spans":[],"droppedSpanCount":0}"#; + let error = replay_trace( + LocallyReviewedTraceArtifact::new(reviewed).expect("reviewed artifact"), + consent(), + &CancellationToken::new(), + ) + .expect_err("cross-tool concurrency must fail"); + assert_eq!(error, TraceReplayError::Busy); + + drop(held); + replay_trace( + LocallyReviewedTraceArtifact::new(reviewed).expect("reviewed artifact"), + consent(), + &CancellationToken::new(), + ) + .expect("replay succeeds after lease release"); +} + +#[tokio::test] +#[serial] +async fn connect_trace_otlp_honors_stop_without_contacting_the_collector() { + let cancel = CancellationToken::new(); + cancel.cancel(); + let error = export_trace_otlp( + Url::parse("http://127.0.0.1:9/v1/traces").expect("URL"), + LocalOtlpHeaders::new(header::HeaderMap::new()), + OtlpBatch::new(encoded_batch(1)).expect("valid batch"), + consent(), + Duration::from_secs(1), + &cancel, + ) + .await + .expect_err("cancelled forward must fail"); + assert_eq!(error, OtlpForwardError::Cancelled); +} + +#[tokio::test] +#[serial] +async fn connect_trace_otlp_enforces_the_timeout_byte_budget_before_network_io() { + let endpoint = Url::parse("http://127.0.0.1:9/v1/traces").expect("URL"); + let error = export_trace_otlp( + endpoint.clone(), + LocalOtlpHeaders::new(header::HeaderMap::new()), + OtlpBatch::new(encoded_batch(1024)).expect("bounded actual batch"), + consent(), + Duration::from_millis(1), + &CancellationToken::new(), + ) + .await + .expect_err("one millisecond cannot send the full batch budget"); + assert_eq!(error, OtlpForwardError::InvalidBatch); + + let error = export_trace_otlp( + endpoint, + LocalOtlpHeaders::new(header::HeaderMap::new()), + OtlpBatch::new(encoded_batch(1)).expect("bounded actual batch"), + consent(), + Duration::from_micros(999), + &CancellationToken::new(), + ) + .await + .expect_err("sub-millisecond timeout has no byte budget"); + assert_eq!(error, OtlpForwardError::InvalidTimeout); + + let error = export_trace_otlp( + Url::parse("http://127.0.0.1:9/v1/traces").expect("URL"), + LocalOtlpHeaders::new(header::HeaderMap::new()), + OtlpBatch::new(encoded_named_batch(200_000)).expect("bounded actual batch"), + LocalTelemetryConsent::new(Instant::now() + Duration::from_millis(100)).expect("short consent"), + Duration::from_secs(30), + &CancellationToken::new(), + ) + .await + .expect_err("effective consent timeout limits the byte budget"); + assert_eq!(error, OtlpForwardError::InvalidBatch); +} diff --git a/rustfs/tests/connect_trace_record.rs b/rustfs/tests/connect_trace_record.rs new file mode 100644 index 000000000..85bc5553b --- /dev/null +++ b/rustfs/tests/connect_trace_record.rs @@ -0,0 +1,440 @@ +use serial_test::serial; +use std::io::{Cursor, Read as _}; +#[cfg(unix)] +use std::os::unix::fs::PermissionsExt as _; +use std::time::{Duration, Instant, 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 rustfs::connect::{ + DeviceIdentity, LocalTelemetryConsent, ObservedTelemetrySpan, TelemetryArtifactConsent, TelemetryArtifactError, + TelemetryArtifactRequest, TelemetryDiagnosticResult, TelemetryOperation, TelemetryProducerError, TelemetryProvenance, + TelemetrySpanStatus, TelemetryTool, TraceRecordCompletion, TraceRecordLimits, encode_signed_telemetry_export, + record_diagnostic_result, record_trace, record_trace_bus, save_signed_telemetry_export, +}; +use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit}; +use sha2::{Digest as _, Sha256}; +use tokio::sync::mpsc; +use tokio_util::sync::CancellationToken; +use zip::ZipArchive; + +fn consent() -> LocalTelemetryConsent { + LocalTelemetryConsent::new(Instant::now() + Duration::from_secs(5)).expect("future consent") +} + +fn artifact_request() -> TelemetryArtifactRequest { + let now = SystemTime::now().duration_since(UNIX_EPOCH).expect("current time").as_secs() as i64; + let organization = "organizations/019e3ae0-0000-7000-8000-000000000001"; + let cluster = format!("{organization}/clusters/019e3ae0-0000-7000-8000-000000000002"); + TelemetryArtifactRequest { + organization_name: organization.to_owned(), + cluster_name: cluster.clone(), + device_name: format!("{cluster}/clusterDevices/019e3ae0-0000-7000-8000-000000000003"), + run_uid: "019e3ae0-0000-7000-8000-000000000004".to_owned(), + artifact_uid: "019e3ae0-0000-7000-8000-000000000005".to_owned(), + schema_version: 1, + consent: TelemetryArtifactConsent { + consent_uid: "019e3ae0-0000-7000-8000-000000000006".to_owned(), + policy_revision: 1, + expires_at_unix: now + 120, + confirmed: true, + }, + produced_at_unix: now, + expires_at_unix: now + 60, + nonce: [0x5a; 32], + provenance: TelemetryProvenance::new("a".repeat(40), "b".repeat(64), "1.0.0-rc.6", vec![]), + } +} + +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] +#[serial] +async fn connect_trace_record_emits_only_the_frozen_redacted_shape() { + let (sender, receiver) = mpsc::channel(4); + sender + .send(ObservedTelemetrySpan::new( + TelemetryOperation::GetObject, + Duration::from_micros(500), + TelemetrySpanStatus::Ok, + )) + .await + .expect("open source"); + let task = tokio::spawn(async move { + record_trace( + receiver, + consent(), + TraceRecordLimits { + duration: Duration::from_millis(15), + max_spans: 4, + }, + &CancellationToken::new(), + ) + .await + }); + tokio::time::sleep(Duration::from_millis(20)).await; + let record = task.await.expect("capture task").expect("capture succeeds"); + assert_eq!(record.completion, TraceRecordCompletion::Complete); + let json = serde_json::to_value(&record.data).expect("serialize record"); + + assert_eq!(json["spans"][0]["operation"], "GET_OBJECT"); + assert_eq!(json["spans"][0]["durationMicros"], 500); + assert_eq!(json["spans"][0]["status"], "OK"); + assert_eq!(json["droppedSpanCount"], 0); + let mut root_keys = json + .as_object() + .expect("record object") + .keys() + .map(String::as_str) + .collect::>(); + let mut span_keys = json["spans"][0] + .as_object() + .expect("span object") + .keys() + .map(String::as_str) + .collect::>(); + root_keys.sort_unstable(); + span_keys.sort_unstable(); + assert_eq!(root_keys, ["droppedSpanCount", "spans"]); + assert_eq!(span_keys, ["durationMicros", "operation", "status"]); +} + +#[tokio::test] +#[serial] +async fn connect_trace_record_stops_at_span_limit_and_counts_queued_drops() { + let (sender, receiver) = mpsc::channel(8); + for _ in 0..3 { + sender + .send(ObservedTelemetrySpan::new( + TelemetryOperation::InternalRpc, + Duration::from_micros(1), + TelemetrySpanStatus::Error, + )) + .await + .expect("open source"); + } + let record = record_trace( + receiver, + consent(), + TraceRecordLimits { + duration: Duration::from_secs(1), + max_spans: 1, + }, + &CancellationToken::new(), + ) + .await + .expect("bounded capture succeeds"); + + assert_eq!(record.data.spans.len(), 1); + assert_eq!(record.data.dropped_span_count, 2); + assert_eq!(record.completion, TraceRecordCompletion::LimitExceeded); +} + +#[tokio::test] +#[serial] +async fn connect_trace_record_marks_a_closed_source_incomplete_after_real_observations() { + let (sender, receiver) = mpsc::channel(1); + sender + .send(ObservedTelemetrySpan::new( + TelemetryOperation::HeadObject, + Duration::from_micros(2), + TelemetrySpanStatus::Ok, + )) + .await + .expect("open source"); + drop(sender); + let record = record_trace( + receiver, + consent(), + TraceRecordLimits { + duration: Duration::from_secs(1), + max_spans: 2, + }, + &CancellationToken::new(), + ) + .await + .expect("observed data remains usable"); + + assert_eq!(record.data.spans.len(), 1); + assert_eq!(record.completion, TraceRecordCompletion::SourceUnavailable); +} + +#[tokio::test] +#[serial] +async fn connect_trace_record_rejects_limits_and_honors_stop() { + let (_sender, receiver) = mpsc::channel(1); + let error = record_trace( + receiver, + consent(), + TraceRecordLimits { + duration: Duration::from_secs(31), + max_spans: 1, + }, + &CancellationToken::new(), + ) + .await + .expect_err("overlong capture must fail"); + assert_eq!(error, TelemetryProducerError::InvalidDuration); + + let (_sender, receiver) = mpsc::channel(1); + let cancel = CancellationToken::new(); + cancel.cancel(); + let error = record_trace( + receiver, + consent(), + TraceRecordLimits { + duration: Duration::from_secs(1), + max_spans: 1, + }, + &cancel, + ) + .await + .expect_err("cancelled capture must fail"); + assert_eq!(error, TelemetryProducerError::Cancelled); +} + +#[test] +#[serial] +fn connect_trace_record_requires_live_local_consent() { + assert_eq!( + LocalTelemetryConsent::new(Instant::now()).expect_err("expired consent must fail"), + TelemetryProducerError::ConsentExpired + ); +} + +#[test] +#[serial] +fn connect_trace_record_enforces_exact_duration_and_sample_boundaries() { + assert!( + TraceRecordLimits { + duration: Duration::from_secs(30), + max_spans: 1024, + } + .validate() + .is_ok() + ); + assert_eq!( + TraceRecordLimits { + duration: Duration::from_secs(30) + Duration::from_nanos(1), + max_spans: 1024, + } + .validate() + .expect_err("duration N+1 must fail"), + TelemetryProducerError::InvalidDuration + ); + assert_eq!( + TraceRecordLimits { + duration: Duration::from_secs(30), + max_spans: 1025, + } + .validate() + .expect_err("sample N+1 must fail"), + TelemetryProducerError::InvalidSpanLimit + ); +} + +#[tokio::test] +#[serial] +async fn connect_trace_record_refuses_to_relabel_the_real_heal_bus_as_frozen_telemetry() { + let task = tokio::spawn(async { + record_trace_bus( + consent(), + TraceRecordLimits { + duration: Duration::from_millis(80), + max_spans: 8, + }, + &CancellationToken::new(), + ) + .await + }); + tokio::time::sleep(Duration::from_millis(10)).await; + assert!(trace_emit(|| { + TraceEvent::new(TraceKind::Scanner, TraceFunc::ScannerHealCandidate) + .with_bucket("SYNTHETIC_SECRET_BUCKET") + .with_object("private/object") + .with_duration(Duration::from_micros(41)) + .with_attr("error", "SYNTHETIC_SECRET_ERROR") + })); + let error = task + .await + .expect("capture task") + .expect_err("heal/scanner events have no frozen telemetry semantics"); + assert_eq!(error, TelemetryProducerError::SourceUnavailable); +} + +#[test] +#[serial] +fn connect_trace_record_writes_a_verified_no_clobber_signed_archive() { + let request = artifact_request(); + let key = DeviceIdentity::generate(); + let result = record_diagnostic_result( + &request, + rustfs::connect::TraceRecordCapture { + data: rustfs::connect::RecordedTrace { + spans: vec![rustfs::connect::TelemetrySpan { + operation: TelemetryOperation::GetObject, + duration_micros: 500, + status: TelemetrySpanStatus::Ok, + }], + dropped_span_count: 0, + }, + completion: TraceRecordCompletion::Complete, + }, + Duration::from_millis(1), + ); + let export = + encode_signed_telemetry_export(&request, &result, &key, &CancellationToken::new()).expect("signed telemetry export"); + let mut archive = ZipArchive::new(Cursor::new(export.archive_bytes.clone())).expect("telemetry 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_json: serde_json::Value = serde_json::from_slice(&result_bytes).expect("result JSON"); + assert_eq!(envelope["formatVersion"], "rustfs.connect.diagnosticEnvelope/1"); + assert_eq!(envelope["toolId"], "telemetry.record"); + assert_eq!(envelope["classification"], "L3"); + assert_eq!( + envelope["payload"]["sha256"], + hex_simd::encode_to_string(Sha256::digest(&result_bytes), hex_simd::AsciiCase::Lower) + ); + assert_eq!(result_json["toolId"], "telemetry.record"); + assert_eq!(result_json["capability"], "telemetry.record@1"); + assert_eq!(result_json["outcome"], "SUCCEEDED"); + assert_eq!(result_json["provenance"]["sourceCommit"], "a".repeat(40)); + + let raw_signature = URL_SAFE_NO_PAD + .decode_to_vec(signature["value"].as_str().expect("signature value")) + .expect("base64url signature"); + let signature = Signature::from_slice(&raw_signature).expect("P-256 signature"); + 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).expect("valid envelope signature"); + + let directory = tempfile::tempdir().expect("temporary output"); + let output = directory.path().join("telemetry.zip"); + save_signed_telemetry_export(&output, &export, &CancellationToken::new()).expect("saved telemetry export"); + #[cfg(unix)] + assert_eq!(std::fs::metadata(&output).expect("output metadata").permissions().mode() & 0o777, 0o600); + assert!(matches!( + save_signed_telemetry_export(&output, &export, &CancellationToken::new()), + Err(TelemetryArtifactError::AlreadyExists) + )); +} + +#[test] +#[serial] +fn connect_trace_record_rejects_tampered_or_cancelled_archives_without_a_file() { + let request = artifact_request(); + let key = DeviceIdentity::generate(); + let result = TelemetryDiagnosticResult::succeeded( + &request, + TelemetryTool::Record, + Duration::ZERO, + rustfs::connect::RecordedTrace { + spans: vec![], + dropped_span_count: 0, + }, + ); + let export = + encode_signed_telemetry_export(&request, &result, &key, &CancellationToken::new()).expect("signed telemetry export"); + let directory = tempfile::tempdir().expect("temporary output"); + + let tampered_output = directory.path().join("tampered.zip"); + let mut tampered = export.clone(); + tampered.archive_bytes.push(0); + assert!(matches!( + save_signed_telemetry_export(&tampered_output, &tampered, &CancellationToken::new()), + Err(TelemetryArtifactError::InvalidRequest) + )); + assert!(!tampered_output.exists()); + + let cancelled_output = directory.path().join("cancelled.zip"); + let cancel = CancellationToken::new(); + cancel.cancel(); + assert!(matches!( + save_signed_telemetry_export(&cancelled_output, &export, &cancel), + Err(TelemetryArtifactError::Cancelled) + )); + assert!(!cancelled_output.exists()); +} + +#[test] +#[serial] +fn connect_trace_record_never_publishes_an_unsupported_artifact() { + let request = artifact_request(); + let key = DeviceIdentity::generate(); + let result = + TelemetryDiagnosticResult::::unsupported(&request, TelemetryTool::Record, Duration::ZERO); + assert!(matches!( + encode_signed_telemetry_export(&request, &result, &key, &CancellationToken::new()), + Err(TelemetryArtifactError::InvalidRequest) + )); +} + +#[test] +#[serial] +fn connect_trace_record_never_publishes_an_incomplete_single_window() { + let request = artifact_request(); + let key = DeviceIdentity::generate(); + for (completion, expected_reason) in [ + (TraceRecordCompletion::LimitExceeded, "COLLECTION_FAILED"), + (TraceRecordCompletion::SourceUnavailable, "SOURCE_UNAVAILABLE"), + ] { + let result = record_diagnostic_result( + &request, + rustfs::connect::TraceRecordCapture { + data: rustfs::connect::RecordedTrace { + spans: vec![rustfs::connect::TelemetrySpan { + operation: TelemetryOperation::GetObject, + duration_micros: 500, + status: TelemetrySpanStatus::Ok, + }], + dropped_span_count: 1, + }, + completion, + }, + Duration::from_millis(1), + ); + let serialized = serde_json::to_value(&result).expect("failed diagnostic result"); + assert_eq!(serialized["outcome"], "FAILED"); + assert_eq!(serialized["coverage"]["requestedUnits"], 1); + assert_eq!(serialized["coverage"]["completedUnits"], 0); + assert_eq!(serialized["reasonCode"], expected_reason); + assert!(serialized["data"].is_null()); + assert!(matches!( + encode_signed_telemetry_export(&request, &result, &key, &CancellationToken::new()), + Err(TelemetryArtifactError::InvalidRequest) + )); + } +} + +#[test] +#[serial] +fn connect_trace_record_rejects_a_forged_partial_complete_result() { + let request = artifact_request(); + let key = DeviceIdentity::generate(); + let result = TelemetryDiagnosticResult::partial( + &request, + TelemetryTool::Record, + Duration::ZERO, + rustfs::connect::TelemetryReasonCode::Complete, + rustfs::connect::RecordedTrace { + spans: vec![], + dropped_span_count: 0, + }, + ); + assert!(matches!( + encode_signed_telemetry_export(&request, &result, &key, &CancellationToken::new()), + Err(TelemetryArtifactError::InvalidRequest) + )); +} diff --git a/rustfs/tests/connect_trace_replay.rs b/rustfs/tests/connect_trace_replay.rs new file mode 100644 index 000000000..bf0416d1c --- /dev/null +++ b/rustfs/tests/connect_trace_replay.rs @@ -0,0 +1,135 @@ +use serial_test::serial; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use rustfs::connect::{ + LocalTelemetryConsent, LocallyReviewedTraceArtifact, TelemetryArtifactConsent, TelemetryArtifactRequest, TelemetryProvenance, + TraceReplayError, analyze_trace, replay_trace, replay_trace_result, +}; +use tokio_util::sync::CancellationToken; + +fn consent() -> LocalTelemetryConsent { + LocalTelemetryConsent::new(Instant::now() + Duration::from_secs(1)).expect("future consent") +} + +fn artifact_request() -> TelemetryArtifactRequest { + let now = SystemTime::now().duration_since(UNIX_EPOCH).expect("current time").as_secs() as i64; + let organization = "organizations/019e3ae0-0000-7000-8000-000000000001"; + let cluster = format!("{organization}/clusters/019e3ae0-0000-7000-8000-000000000002"); + TelemetryArtifactRequest { + organization_name: organization.to_owned(), + cluster_name: cluster.clone(), + device_name: format!("{cluster}/clusterDevices/019e3ae0-0000-7000-8000-000000000003"), + run_uid: "019e3ae0-0000-7000-8000-000000000004".to_owned(), + artifact_uid: "019e3ae0-0000-7000-8000-000000000005".to_owned(), + schema_version: 1, + consent: TelemetryArtifactConsent { + consent_uid: "019e3ae0-0000-7000-8000-000000000006".to_owned(), + policy_revision: 1, + expires_at_unix: now + 120, + confirmed: true, + }, + produced_at_unix: now, + expires_at_unix: now + 60, + nonce: [0x5a; 32], + provenance: TelemetryProvenance::new("a".repeat(40), "b".repeat(64), "1.0.0-rc.6", vec![]), + } +} + +#[test] +#[serial] +fn connect_trace_replay_validates_and_digests_reviewed_bytes_without_file_io() { + let bytes = br#"{"spans":[{"operation":"GET_OBJECT","durationMicros":500,"status":"OK"}],"droppedSpanCount":0}"#; + let replay = replay_trace( + LocallyReviewedTraceArtifact::new(bytes).expect("bounded artifact"), + consent(), + &CancellationToken::new(), + ) + .expect("valid record replays"); + + assert_eq!(replay.spans.len(), 1); + assert_eq!( + replay.input_artifact_sha256, + "1daadd3e226e7dd67a35dd56092ec985f464d832756e75277bc11d4a25c0291b" + ); +} + +#[test] +#[serial] +fn connect_trace_replay_wraps_reviewed_bytes_in_the_frozen_result_contract() { + let bytes = br#"{"spans":[{"operation":"GET_OBJECT","durationMicros":500,"status":"OK"}],"droppedSpanCount":0}"#; + let result = replay_trace_result( + &artifact_request(), + LocallyReviewedTraceArtifact::new(bytes).expect("bounded artifact"), + consent(), + &CancellationToken::new(), + ) + .expect("typed replay result"); + let analysis = analyze_trace(result.data().expect("replayed data"), consent(), &CancellationToken::new()) + .expect("replayed trace is locally analyzable"); + assert_eq!(analysis.span_count, 1); + assert_eq!(analysis.total_duration_micros, 500); + let json = serde_json::to_value(result).expect("result JSON"); + assert_eq!(json["toolId"], "telemetry.replay"); + assert_eq!(json["capability"], "telemetry.replay@1"); + assert_eq!(json["outcome"], "SUCCEEDED"); + assert_eq!(json["reasonCode"], "COMPLETE"); + assert_eq!(json["coverage"]["requestedUnits"], 1); + assert_eq!(json["coverage"]["completedUnits"], 1); + assert_eq!(json["data"]["spans"][0]["operation"], "GET_OBJECT"); +} + +#[test] +#[serial] +fn connect_trace_replay_rejects_unknown_or_secret_bearing_fields() { + let bytes = br#"{"spans":[],"droppedSpanCount":0,"authorization":"Bearer secret"}"#; + let error = replay_trace( + LocallyReviewedTraceArtifact::new(bytes).expect("bounded artifact"), + consent(), + &CancellationToken::new(), + ) + .expect_err("unknown field must fail"); + assert_eq!(error, TraceReplayError::InvalidArtifact); +} + +#[test] +#[serial] +fn connect_trace_replay_honors_stop_before_parsing() { + let cancel = CancellationToken::new(); + cancel.cancel(); + let error = replay_trace( + LocallyReviewedTraceArtifact::new(br#"{"spans":[],"droppedSpanCount":0}"#).expect("bounded artifact"), + consent(), + &cancel, + ) + .expect_err("cancelled replay must fail"); + assert_eq!(error, TraceReplayError::Cancelled); +} + +#[test] +#[serial] +fn connect_trace_replay_rejects_duplicate_fields_and_safe_integer_overflow() { + for bytes in [ + br#"{"spans":[],"spans":[],"droppedSpanCount":0}"#.as_slice(), + br#"{"spans":[{"operation":"GET_OBJECT","durationMicros":9007199254740992,"status":"OK"}],"droppedSpanCount":0}"# + .as_slice(), + br#"{"spans":[],"droppedSpanCount":9007199254740992}"#.as_slice(), + ] { + let error = replay_trace( + LocallyReviewedTraceArtifact::new(bytes).expect("bounded artifact"), + consent(), + &CancellationToken::new(), + ) + .expect_err("invalid record must fail"); + assert_eq!(error, TraceReplayError::InvalidArtifact); + } +} + +#[test] +#[serial] +fn connect_trace_replay_enforces_exact_artifact_size_boundary() { + assert!(LocallyReviewedTraceArtifact::new(&vec![b' '; 262_144]).is_ok()); + assert_eq!( + LocallyReviewedTraceArtifact::new(&vec![b' '; 262_145]).err(), + Some(TraceReplayError::InvalidArtifact) + ); +}