diff --git a/rustfs/src/connect/diagnostics/mod.rs b/rustfs/src/connect/diagnostics/mod.rs index 276392ca5..8a2f53be1 100644 --- a/rustfs/src/connect/diagnostics/mod.rs +++ b/rustfs/src/connect/diagnostics/mod.rs @@ -33,6 +33,11 @@ mod trace_analysis; mod trace_otlp; mod trace_record; mod trace_replay; +#[cfg(unix)] +mod trace_runtime; +#[cfg(not(unix))] +#[path = "trace_runtime_unsupported.rs"] +mod trace_runtime; pub use inspect::{ INSPECT_CAPABILITY, INSPECT_SCHEMA_VERSION, InspectArtifactConsent, InspectDiagnosticResult, InspectError, InspectFinding, @@ -123,3 +128,6 @@ pub use trace_record::{ record_diagnostic_result, record_trace, record_trace_bus, save_signed_telemetry_export, }; pub use trace_replay::{LocallyReviewedTraceArtifact, ReplayedTrace, TraceReplayError, replay_trace, replay_trace_result}; +pub(crate) use trace_runtime::{ + LocalTraceCaptureError, LocalTraceCaptureRuntime, request_local_trace_capture, spawn_local_trace_capture_runtime, +}; diff --git a/rustfs/src/connect/diagnostics/trace_record.rs b/rustfs/src/connect/diagnostics/trace_record.rs index 523414609..5478beab5 100644 --- a/rustfs/src/connect/diagnostics/trace_record.rs +++ b/rustfs/src/connect/diagnostics/trace_record.rs @@ -133,14 +133,14 @@ pub struct RecordedTrace { pub dropped_span_count: u64, } -#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] pub enum TraceRecordCompletion { Complete, LimitExceeded, SourceUnavailable, } -#[derive(Clone, Debug, Eq, PartialEq)] +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] pub struct TraceRecordCapture { pub data: RecordedTrace, pub completion: TraceRecordCompletion, diff --git a/rustfs/src/connect/diagnostics/trace_runtime.rs b/rustfs/src/connect/diagnostics/trace_runtime.rs new file mode 100644 index 000000000..a9418338f --- /dev/null +++ b/rustfs/src/connect/diagnostics/trace_runtime.rs @@ -0,0 +1,526 @@ +// 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. + +//! Owner-only local transport between the telemetry CLI and the running server. + +use std::io; +use std::os::unix::fs::{FileTypeExt as _, MetadataExt as _, PermissionsExt as _}; +use std::path::Path; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use serde::{Deserialize, Serialize}; +use thiserror::Error; +use tokio::io::{AsyncBufReadExt as _, AsyncReadExt as _, AsyncWriteExt as _, BufReader}; +use tokio::net::{UnixListener, UnixStream}; +use tokio::task::{JoinHandle, JoinSet}; +use tokio_util::sync::CancellationToken; + +use super::{LocalTelemetryConsent, TelemetryProducerError, TraceRecordCapture, TraceRecordLimits, record_trace_bus}; + +const SOCKET_FILE: &str = "telemetry-record.sock"; +const PROTOCOL_VERSION: u16 = 1; +const MAX_REQUEST_BYTES: u64 = 1_024; +const MAX_RESPONSE_BYTES: u64 = 300_000; +const MAX_CONNECTIONS: usize = 8; +const REQUEST_TIMEOUT: Duration = Duration::from_secs(1); +const STATE_DIRECTORY_MODE: u32 = 0o700; +const SOCKET_MODE: u32 = 0o600; + +#[derive(Debug, Error)] +pub(crate) enum LocalTraceCaptureError { + #[error("telemetry server runtime is unavailable")] + RuntimeUnavailable, + #[error("telemetry server runtime state is not owner-only")] + StateSecurity, + #[error("telemetry server runtime protocol failed")] + Protocol, + #[error("telemetry server runtime I/O failed")] + Io(#[source] io::Error), + #[error(transparent)] + Producer(#[from] TelemetryProducerError), +} + +pub(crate) struct LocalTraceCaptureRuntime { + shutdown: CancellationToken, + task: Option>, +} + +impl LocalTraceCaptureRuntime { + pub async fn shutdown(mut self) { + self.shutdown.cancel(); + if let Some(task) = self.task.take() { + let _ = task.await; + } + } +} + +impl Drop for LocalTraceCaptureRuntime { + fn drop(&mut self) { + self.shutdown.cancel(); + } +} + +#[derive(Debug, Deserialize, Serialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +struct CaptureRequest { + protocol_version: u16, + consent_expires_at_unix: i64, + duration_millis: u64, + max_spans: usize, +} + +#[derive(Clone, Copy, Debug, Deserialize, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +enum ProducerErrorCode { + Busy, + ConsentExpired, + InvalidDuration, + InvalidSpanLimit, + Cancelled, + SourceUnavailable, + DurationOverflow, + ResultTooLarge, +} + +impl From<&TelemetryProducerError> for ProducerErrorCode { + fn from(error: &TelemetryProducerError) -> Self { + match error { + TelemetryProducerError::Busy => Self::Busy, + TelemetryProducerError::ConsentExpired => Self::ConsentExpired, + TelemetryProducerError::InvalidDuration => Self::InvalidDuration, + TelemetryProducerError::InvalidSpanLimit => Self::InvalidSpanLimit, + TelemetryProducerError::Cancelled => Self::Cancelled, + TelemetryProducerError::SourceUnavailable => Self::SourceUnavailable, + TelemetryProducerError::DurationOverflow => Self::DurationOverflow, + TelemetryProducerError::ResultTooLarge => Self::ResultTooLarge, + } + } +} + +impl From for TelemetryProducerError { + fn from(code: ProducerErrorCode) -> Self { + match code { + ProducerErrorCode::Busy => Self::Busy, + ProducerErrorCode::ConsentExpired => Self::ConsentExpired, + ProducerErrorCode::InvalidDuration => Self::InvalidDuration, + ProducerErrorCode::InvalidSpanLimit => Self::InvalidSpanLimit, + ProducerErrorCode::Cancelled => Self::Cancelled, + ProducerErrorCode::SourceUnavailable => Self::SourceUnavailable, + ProducerErrorCode::DurationOverflow => Self::DurationOverflow, + ProducerErrorCode::ResultTooLarge => Self::ResultTooLarge, + } + } +} + +#[derive(Debug, Deserialize, Serialize)] +#[serde(deny_unknown_fields, tag = "status", rename_all = "SCREAMING_SNAKE_CASE")] +enum CaptureResponse { + Ok { capture: TraceRecordCapture }, + Error { code: ProducerErrorCode }, +} + +pub(crate) fn spawn_local_trace_capture_runtime( + state_root: &Path, + parent_shutdown: &CancellationToken, +) -> Result { + let owner = private_state_owner(state_root)?; + let socket_path = state_root.join(SOCKET_FILE); + let listener = bind_listener(&socket_path, owner)?; + let socket_identity = socket_identity(&socket_path, owner)?; + let shutdown = parent_shutdown.child_token(); + let task_shutdown = shutdown.clone(); + let task = tokio::spawn(async move { + run_listener(listener, owner, task_shutdown).await; + remove_own_socket(&socket_path, socket_identity); + }); + Ok(LocalTraceCaptureRuntime { + shutdown, + task: Some(task), + }) +} + +pub(crate) async fn request_local_trace_capture( + state_root: &Path, + consent_expires_at_unix: i64, + limits: TraceRecordLimits, + cancel: &CancellationToken, +) -> Result { + limits.validate()?; + let owner = private_state_owner(state_root)?; + let socket_path = state_root.join(SOCKET_FILE); + socket_identity(&socket_path, owner)?; + let stream = UnixStream::connect(&socket_path).await.map_err(|error| match error.kind() { + io::ErrorKind::NotFound | io::ErrorKind::ConnectionRefused => LocalTraceCaptureError::RuntimeUnavailable, + _ => LocalTraceCaptureError::Io(error), + })?; + let (reader, mut writer) = stream.into_split(); + let request = CaptureRequest { + protocol_version: PROTOCOL_VERSION, + consent_expires_at_unix, + duration_millis: u64::try_from(limits.duration.as_millis()).map_err(|_| TelemetryProducerError::InvalidDuration)?, + max_spans: limits.max_spans, + }; + let mut encoded = serde_json::to_vec(&request).map_err(|_| LocalTraceCaptureError::Protocol)?; + encoded.push(b'\n'); + if encoded.len() as u64 > MAX_REQUEST_BYTES { + return Err(LocalTraceCaptureError::Protocol); + } + writer.write_all(&encoded).await.map_err(LocalTraceCaptureError::Io)?; + + let response = async move { + let mut bytes = Vec::new(); + reader + .take(MAX_RESPONSE_BYTES + 1) + .read_to_end(&mut bytes) + .await + .map_err(LocalTraceCaptureError::Io)?; + if bytes.is_empty() || bytes.len() as u64 > MAX_RESPONSE_BYTES { + return Err(LocalTraceCaptureError::Protocol); + } + serde_json::from_slice::(&bytes).map_err(|_| LocalTraceCaptureError::Protocol) + }; + let response = tokio::select! { + biased; + _ = cancel.cancelled() => return Err(TelemetryProducerError::Cancelled.into()), + response = response => response?, + }; + match response { + CaptureResponse::Ok { capture } => Ok(capture), + CaptureResponse::Error { code } => Err(TelemetryProducerError::from(code).into()), + } +} + +async fn run_listener(listener: UnixListener, owner: u32, shutdown: CancellationToken) { + let mut connections = JoinSet::new(); + loop { + tokio::select! { + biased; + _ = shutdown.cancelled() => break, + Some(_) = connections.join_next(), if !connections.is_empty() => {} + accepted = listener.accept() => match accepted { + Ok((stream, _)) => { + if connections.len() < MAX_CONNECTIONS + && stream.peer_cred().is_ok_and(|credentials| credentials.uid() == owner) + { + connections.spawn(handle_connection(stream, shutdown.clone())); + } + } + Err(_) => break, + }, + } + } + while connections.join_next().await.is_some() {} +} + +async fn handle_connection(stream: UnixStream, shutdown: CancellationToken) { + let (reader, mut writer) = stream.into_split(); + let mut reader = BufReader::new(reader); + let request = tokio::select! { + biased; + _ = shutdown.cancelled() => return, + result = tokio::time::timeout(REQUEST_TIMEOUT, read_request(&mut reader)) => match result { + Ok(Ok(request)) => request, + Ok(Err(_)) | Err(_) => return, + }, + }; + let limits = TraceRecordLimits { + duration: Duration::from_millis(request.duration_millis), + max_spans: request.max_spans, + }; + let capture = async { + if request.protocol_version != PROTOCOL_VERSION { + return Err(TelemetryProducerError::SourceUnavailable); + } + limits.validate()?; + let remaining = request + .consent_expires_at_unix + .checked_sub(unix_now().map_err(|_| TelemetryProducerError::ConsentExpired)?) + .and_then(|seconds| u64::try_from(seconds).ok()) + .filter(|seconds| *seconds > 0) + .ok_or(TelemetryProducerError::ConsentExpired)?; + let expires_at = Instant::now() + .checked_add(Duration::from_secs(remaining)) + .ok_or(TelemetryProducerError::ConsentExpired)?; + let consent = LocalTelemetryConsent::new(expires_at)?; + record_trace_bus(consent, limits, &shutdown).await + }; + tokio::pin!(capture); + let mut unexpected = [0_u8; 1]; + let disconnected = reader.read(&mut unexpected); + tokio::pin!(disconnected); + let result = tokio::select! { + biased; + _ = shutdown.cancelled() => Err(TelemetryProducerError::Cancelled), + _ = &mut disconnected => return, + result = &mut capture => result, + }; + let response = match result { + Ok(capture) => CaptureResponse::Ok { capture }, + Err(error) => CaptureResponse::Error { + code: ProducerErrorCode::from(&error), + }, + }; + if let Ok(bytes) = serde_json::to_vec(&response) { + let write_response = async { + writer.write_all(&bytes).await?; + writer.shutdown().await + }; + tokio::select! { + biased; + _ = shutdown.cancelled() => {} + _ = tokio::time::timeout(REQUEST_TIMEOUT, write_response) => {} + } + } +} + +async fn read_request(reader: &mut BufReader) -> Result { + let mut bytes = Vec::new(); + reader + .take(MAX_REQUEST_BYTES + 1) + .read_until(b'\n', &mut bytes) + .await + .map_err(LocalTraceCaptureError::Io)?; + if bytes.is_empty() || bytes.len() as u64 > MAX_REQUEST_BYTES || bytes.pop() != Some(b'\n') { + return Err(LocalTraceCaptureError::Protocol); + } + serde_json::from_slice(&bytes).map_err(|_| LocalTraceCaptureError::Protocol) +} + +fn private_state_owner(path: &Path) -> Result { + let metadata = std::fs::symlink_metadata(path).map_err(LocalTraceCaptureError::Io)?; + if metadata.file_type().is_symlink() || !metadata.is_dir() || metadata.permissions().mode() & 0o777 != STATE_DIRECTORY_MODE { + return Err(LocalTraceCaptureError::StateSecurity); + } + Ok(metadata.uid()) +} + +fn bind_listener(path: &Path, owner: u32) -> Result { + match UnixListener::bind(path) { + Ok(listener) => seal_listener(path, owner, listener), + Err(error) if error.kind() == io::ErrorKind::AddrInUse => { + let identity = socket_identity(path, owner)?; + match std::os::unix::net::UnixStream::connect(path) { + Ok(_) => Err(LocalTraceCaptureError::RuntimeUnavailable), + Err(connect_error) if connect_error.kind() == io::ErrorKind::ConnectionRefused => { + remove_own_socket(path, identity); + let listener = UnixListener::bind(path).map_err(LocalTraceCaptureError::Io)?; + seal_listener(path, owner, listener) + } + Err(connect_error) => Err(LocalTraceCaptureError::Io(connect_error)), + } + } + Err(error) => Err(LocalTraceCaptureError::Io(error)), + } +} + +fn seal_listener(path: &Path, owner: u32, listener: UnixListener) -> Result { + let metadata = std::fs::symlink_metadata(path).map_err(LocalTraceCaptureError::Io)?; + if !metadata.file_type().is_socket() || metadata.uid() != owner { + return Err(LocalTraceCaptureError::StateSecurity); + } + let identity = (metadata.dev(), metadata.ino()); + let sealed = std::fs::set_permissions(path, std::fs::Permissions::from_mode(SOCKET_MODE)) + .map_err(LocalTraceCaptureError::Io) + .and_then(|()| socket_identity(path, owner)); + match sealed { + Ok(sealed_identity) if sealed_identity == identity => Ok(listener), + Ok(_) => Err(LocalTraceCaptureError::StateSecurity), + Err(error) => { + drop(listener); + remove_own_socket(path, identity); + Err(error) + } + } +} + +fn socket_identity(path: &Path, owner: u32) -> Result<(u64, u64), LocalTraceCaptureError> { + let metadata = std::fs::symlink_metadata(path).map_err(|error| match error.kind() { + io::ErrorKind::NotFound => LocalTraceCaptureError::RuntimeUnavailable, + _ => LocalTraceCaptureError::Io(error), + })?; + if !metadata.file_type().is_socket() || metadata.uid() != owner || metadata.permissions().mode() & 0o777 != SOCKET_MODE { + return Err(LocalTraceCaptureError::StateSecurity); + } + Ok((metadata.dev(), metadata.ino())) +} + +fn remove_own_socket(path: &Path, identity: (u64, u64)) { + if std::fs::symlink_metadata(path).is_ok_and(|metadata| (metadata.dev(), metadata.ino()) == identity) { + let _ = std::fs::remove_file(path); + } +} + +fn unix_now() -> io::Result { + let duration = SystemTime::now().duration_since(UNIX_EPOCH).map_err(io::Error::other)?; + i64::try_from(duration.as_secs()).map_err(io::Error::other) +} + +#[cfg(test)] +mod tests { + use std::os::unix::fs::PermissionsExt as _; + use std::time::{Duration, SystemTime, UNIX_EPOCH}; + + use rustfs_common::trace_bus::{ + TelemetryTraceEvent, TelemetryTraceOperation, TelemetryTraceStatus, telemetry_trace_emit, + telemetry_trace_subscriber_count, + }; + use serial_test::serial; + use tokio_util::sync::CancellationToken; + + use super::{LocalTraceCaptureError, request_local_trace_capture, spawn_local_trace_capture_runtime}; + use crate::connect::{TelemetryOperation, TelemetryProducerError, TraceRecordCompletion, TraceRecordLimits}; + + async fn wait_for_subscriber() { + tokio::time::timeout(Duration::from_secs(1), async { + while telemetry_trace_subscriber_count() == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("telemetry subscriber"); + } + + #[tokio::test] + #[serial] + async fn local_runtime_captures_server_process_events() { + let state = tempfile::tempdir().expect("state"); + std::fs::set_permissions(state.path(), std::fs::Permissions::from_mode(0o700)).expect("private state"); + let shutdown = CancellationToken::new(); + let runtime = spawn_local_trace_capture_runtime(state.path(), &shutdown).expect("runtime"); + let request_state = state.path().to_path_buf(); + let consent_expiry = SystemTime::now().duration_since(UNIX_EPOCH).expect("clock").as_secs() as i64 + 5; + let request = tokio::spawn(async move { + request_local_trace_capture( + &request_state, + consent_expiry, + TraceRecordLimits { + duration: Duration::from_millis(40), + max_spans: 8, + }, + &CancellationToken::new(), + ) + .await + }); + wait_for_subscriber().await; + assert!(telemetry_trace_emit(|| TelemetryTraceEvent::new( + TelemetryTraceOperation::GetObject, + Duration::from_micros(7), + TelemetryTraceStatus::Ok, + ))); + let capture = request.await.expect("request task").expect("capture"); + assert_eq!(capture.data.spans.len(), 1); + assert_eq!(capture.data.spans[0].operation, TelemetryOperation::GetObject); + runtime.shutdown().await; + } + + #[tokio::test] + #[serial] + async fn local_runtime_rejects_non_private_state() { + let state = tempfile::tempdir().expect("state"); + std::fs::set_permissions(state.path(), std::fs::Permissions::from_mode(0o755)).expect("public state"); + assert!(spawn_local_trace_capture_runtime(state.path(), &CancellationToken::new()).is_err()); + } + + #[tokio::test] + #[serial] + async fn local_runtime_enforces_consent_and_span_limits() { + let state = tempfile::tempdir().expect("state"); + std::fs::set_permissions(state.path(), std::fs::Permissions::from_mode(0o700)).expect("private state"); + let shutdown = CancellationToken::new(); + let runtime = spawn_local_trace_capture_runtime(state.path(), &shutdown).expect("runtime"); + let now = SystemTime::now().duration_since(UNIX_EPOCH).expect("clock").as_secs() as i64; + + let expired = request_local_trace_capture( + state.path(), + now - 1, + TraceRecordLimits { + duration: Duration::from_millis(10), + max_spans: 1, + }, + &CancellationToken::new(), + ) + .await + .expect_err("expired consent"); + assert!(matches!( + expired, + LocalTraceCaptureError::Producer(TelemetryProducerError::ConsentExpired) + )); + + let request_state = state.path().to_path_buf(); + let request = tokio::spawn(async move { + request_local_trace_capture( + &request_state, + now + 5, + TraceRecordLimits { + duration: Duration::from_secs(1), + max_spans: 1, + }, + &CancellationToken::new(), + ) + .await + }); + wait_for_subscriber().await; + for _ in 0..3 { + assert!(telemetry_trace_emit(|| TelemetryTraceEvent::new( + TelemetryTraceOperation::InternalRpc, + Duration::from_micros(9), + TelemetryTraceStatus::Error, + ))); + } + let capture = request.await.expect("request task").expect("bounded capture"); + assert_eq!(capture.completion, TraceRecordCompletion::LimitExceeded); + assert_eq!(capture.data.spans.len(), 1); + assert_eq!(capture.data.dropped_span_count, 2); + runtime.shutdown().await; + } + + #[tokio::test] + #[serial] + async fn local_runtime_releases_capture_when_the_client_stops() { + let state = tempfile::tempdir().expect("state"); + std::fs::set_permissions(state.path(), std::fs::Permissions::from_mode(0o700)).expect("private state"); + let shutdown = CancellationToken::new(); + let runtime = spawn_local_trace_capture_runtime(state.path(), &shutdown).expect("runtime"); + let now = SystemTime::now().duration_since(UNIX_EPOCH).expect("clock").as_secs() as i64; + let request_state = state.path().to_path_buf(); + let cancel = CancellationToken::new(); + let task_cancel = cancel.clone(); + let request = tokio::spawn(async move { + request_local_trace_capture( + &request_state, + now + 5, + TraceRecordLimits { + duration: Duration::from_secs(1), + max_spans: 8, + }, + &task_cancel, + ) + .await + }); + wait_for_subscriber().await; + cancel.cancel(); + assert!(matches!( + request.await.expect("request task"), + Err(LocalTraceCaptureError::Producer(TelemetryProducerError::Cancelled)) + )); + tokio::time::timeout(Duration::from_secs(1), async { + while telemetry_trace_subscriber_count() != 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("server capture stops after disconnect"); + runtime.shutdown().await; + } +} diff --git a/rustfs/src/connect/diagnostics/trace_runtime_unsupported.rs b/rustfs/src/connect/diagnostics/trace_runtime_unsupported.rs new file mode 100644 index 000000000..af35d7742 --- /dev/null +++ b/rustfs/src/connect/diagnostics/trace_runtime_unsupported.rs @@ -0,0 +1,57 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::io; +use std::path::Path; + +use thiserror::Error; +use tokio_util::sync::CancellationToken; + +use super::{TelemetryProducerError, TraceRecordCapture, TraceRecordLimits}; + +#[derive(Debug, Error)] +pub(crate) enum LocalTraceCaptureError { + #[error("telemetry server runtime is unavailable")] + RuntimeUnavailable, + #[error("telemetry server runtime state is not owner-only")] + StateSecurity, + #[error("telemetry server runtime protocol failed")] + Protocol, + #[error("telemetry server runtime I/O failed")] + Io(#[source] io::Error), + #[error(transparent)] + Producer(#[from] TelemetryProducerError), +} + +pub(crate) struct LocalTraceCaptureRuntime; + +impl LocalTraceCaptureRuntime { + pub async fn shutdown(self) {} +} + +pub(crate) fn spawn_local_trace_capture_runtime( + _state_root: &Path, + _parent_shutdown: &CancellationToken, +) -> Result { + Err(LocalTraceCaptureError::RuntimeUnavailable) +} + +pub(crate) async fn request_local_trace_capture( + _state_root: &Path, + _consent_expires_at_unix: i64, + _limits: TraceRecordLimits, + _cancel: &CancellationToken, +) -> Result { + Err(LocalTraceCaptureError::RuntimeUnavailable) +} diff --git a/rustfs/src/connect/mod.rs b/rustfs/src/connect/mod.rs index c8abdcdfe..d28a1e2f0 100644 --- a/rustfs/src/connect/mod.rs +++ b/rustfs/src/connect/mod.rs @@ -114,6 +114,9 @@ pub use diagnostics::{ TopRpcData, capture_top_api, capture_top_disk, capture_top_locks, capture_top_net, capture_top_rpc, evaluate_disk_window, evaluate_network_window, save_signed_top_export, sign_top_export, }; +pub(crate) use diagnostics::{ + LocalTraceCaptureError, LocalTraceCaptureRuntime, request_local_trace_capture, spawn_local_trace_capture_runtime, +}; pub use environment::{ ENVIRONMENT_CAPABILITY, ENVIRONMENT_SCHEMA_VERSION, EnvironmentCollectionRequest, EnvironmentError, EnvironmentFilesystemType, EnvironmentInventory, EnvironmentOsFamily, MAX_ENVIRONMENT_DURATION, collect_environment, diff --git a/rustfs/src/startup_entrypoint.rs b/rustfs/src/startup_entrypoint.rs index b8ffa6743..54037998b 100644 --- a/rustfs/src/startup_entrypoint.rs +++ b/rustfs/src/startup_entrypoint.rs @@ -302,13 +302,13 @@ 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, + export_trace_otlp_result, record_diagnostic_result, replay_trace_result, request_local_trace_capture, }; use reqwest::header::{AUTHORIZATION, HeaderMap, HeaderValue}; match command { ConnectTelemetryCommands::Record(options) => { - let (key, request, consent) = telemetry_context(&options.artifact)?; + let (key, request, _consent) = telemetry_context(&options.artifact)?; request.validate().map_err(Error::other)?; if options.duration_millis == 0 || options.duration_millis > 30_000 @@ -319,8 +319,9 @@ async fn execute_connect_telemetry(command: ConnectTelemetryCommands) -> Result< } let cancel = CancellationToken::new(); let started = Instant::now(); - let capture = record_trace_bus( - consent, + let capture = request_local_trace_capture( + &options.artifact.state_dir, + options.artifact.consent_expires_at_unix, TraceRecordLimits { duration: Duration::from_millis(options.duration_millis), max_spans: options.max_spans, @@ -342,7 +343,7 @@ async fn execute_connect_telemetry(command: ConnectTelemetryCommands) -> Result< let result = record_diagnostic_result(&request, capture, started.elapsed()); save_telemetry_result(&options.artifact, &request, &result, &key, &cancel, None) } - Err(TelemetryProducerError::SourceUnavailable) => { + Err(crate::connect::LocalTraceCaptureError::Producer(TelemetryProducerError::SourceUnavailable)) => { let result = TelemetryDiagnosticResult::::unsupported( &request, TelemetryTool::Record, diff --git a/rustfs/src/startup_lifecycle.rs b/rustfs/src/startup_lifecycle.rs index 0b19881e0..ed17225ed 100644 --- a/rustfs/src/startup_lifecycle.rs +++ b/rustfs/src/startup_lifecycle.rs @@ -133,6 +133,7 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec optional_runtimes, heartbeat, inventory, + local_trace_capture, iam_bootstrap, enable_scanner, } = service_runtime; @@ -166,6 +167,9 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec ) .await; shutdown_connect_runtimes(heartbeat, inventory).await; + if let Some(runtime) = local_trace_capture { + runtime.shutdown().await; + } if let Some(cleanup) = scanner_cleanup { let _ = wait_for_scanner_cleanup(cleanup).await; } diff --git a/rustfs/src/startup_services.rs b/rustfs/src/startup_services.rs index 2ca351013..1dcee12fb 100644 --- a/rustfs/src/startup_services.rs +++ b/rustfs/src/startup_services.rs @@ -18,8 +18,8 @@ use crate::{ config::Config, connect::{ CoarseNodeSummary, HeartbeatConfig, HeartbeatError, HeartbeatRuntime, InventoryError, InventoryFlag, InventoryRuntime, - InventorySchedule, InventorySnapshot, runtime::heartbeat_failure_reason, spawn_heartbeat_runtime, - spawn_inventory_runtime, + InventorySchedule, InventorySnapshot, LocalTraceCaptureRuntime, runtime::heartbeat_failure_reason, + spawn_heartbeat_runtime, spawn_inventory_runtime, spawn_local_trace_capture_runtime, }, init::{init_buffer_profile_system, init_kms_system}, server::ServiceStateManager, @@ -46,6 +46,7 @@ pub(crate) struct StartupServiceRuntime { pub(crate) optional_runtimes: OptionalRuntimeServices, pub(crate) heartbeat: Option, pub(crate) inventory: Option, + pub(crate) local_trace_capture: Option, pub(crate) iam_bootstrap: IamBootstrapDisposition, pub(crate) enable_scanner: bool, } @@ -85,6 +86,12 @@ pub(crate) async fn init_startup_runtime_services( let optional_runtimes = init_optional_runtime_services().await?; let heartbeat_config = HeartbeatConfig::from_env().map_err(std::io::Error::other)?; + let local_trace_capture = heartbeat_config + .as_ref() + .and_then(HeartbeatConfig::state_root) + .map(|state_root| spawn_local_trace_capture_runtime(state_root, &ctx)) + .transpose() + .map_err(std::io::Error::other)?; let heartbeat_nodes = heartbeat_config.as_ref().map(|_| endpoint_pools.get_nodes().len()); let inventory_drives = heartbeat_config .as_ref() @@ -115,6 +122,7 @@ pub(crate) async fn init_startup_runtime_services( optional_runtimes, heartbeat, inventory, + local_trace_capture, iam_bootstrap, enable_scanner, })