diff --git a/.vscode/launch.json b/.vscode/launch.json index e69583e4a..336254a1b 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -1,7 +1,45 @@ { "version": "0.2.0", "configurations": [ - { + { + "name": "Debug RustFS observability (OTLP)", + "type": "lldb", + "request": "launch", + "cargo": { + "args": [ + "build", + "--bin=rustfs", + "--package=rustfs" + ], + "filter": { + "name": "rustfs", + "kind": "bin" + } + }, + "args": [], + "cwd": "${workspaceFolder}", + "env": { + "RUST_LOG": "rustfs=debug,ecstore=info,s3s=info,iam=info", + "RUST_BACKTRACE": "full", + "RUSTFS_ACCESS_KEY": "rustfsadmin", + "RUSTFS_SECRET_KEY": "rustfsadmin", + "RUSTFS_VOLUMES": "./target/observability/data{1...4}", + "RUSTFS_ADDRESS": ":9000", + "RUSTFS_CONSOLE_ENABLE": "true", + "RUSTFS_CONSOLE_ADDRESS": "127.0.0.1:9001", + "RUSTFS_UNSAFE_BYPASS_DISK_CHECK": "true", + "RUSTFS_OBS_ENDPOINT": "http://127.0.0.1:4318", + "RUSTFS_OBS_TRACES_EXPORT_ENABLED": "true", + "RUSTFS_OBS_METRICS_EXPORT_ENABLED": "true", + "RUSTFS_OBS_LOGS_EXPORT_ENABLED": "true", + "RUSTFS_OBS_USE_STDOUT": "true", + "RUSTFS_OBS_LOG_DIRECTORY": "./target/observability/logs", + "RUSTFS_OBS_METER_INTERVAL": "5", + "RUSTFS_OBS_SERVICE_NAME": "rustfs-observability-local", + "RUSTFS_OBS_ENVIRONMENT": "development" + } + }, + { "type": "lldb", "request": "launch", "name": "Debug(only) executable 'rustfs'", diff --git a/Cargo.lock b/Cargo.lock index 14384120f..fb84ceb82 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9885,6 +9885,7 @@ dependencies = [ "rustfs-data-usage", "rustfs-ecstore", "rustfs-filemeta", + "rustfs-obs", "rustfs-storage-api", "rustfs-utils", "s3s", diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 7053e401d..c82ca95da 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -24,6 +24,7 @@ use super::super::*; use std::future::Future; use std::time::Duration; use tokio::task::JoinSet; +use tracing::Instrument as _; fn map_upload_id_metadata_error(bucket: &str, object: &str, upload_id: &str, err: DiskError) -> Error { if err == DiskError::FileNotFound { @@ -353,6 +354,17 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { let tmp_part = format!("{}x{}", Uuid::new_v4(), OffsetDateTime::now_utc().unix_timestamp()); let tmp_part_path = Arc::new(format!("{tmp_part}/{part_suffix}")); + let ec_write_span = tracing::info_span!( + "rustfs.ec.write", + event = "rustfs_ec_write", + component = "ecstore", + "rustfs.ec.data_shards" = fi.erasure.data_blocks, + "rustfs.ec.parity_shards" = fi.erasure.parity_blocks, + "rustfs.ec.write_quorum" = write_quorum, + "rustfs.ec.block_bytes" = fi.erasure.block_size, + "rustfs.distribution.targets" = shuffle_disks.len(), + "rustfs.distribution.transport" = "erasure" + ); let result: Result = async { let erasure = coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size); let writer_setup_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now); @@ -436,19 +448,29 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { rustfs_io_metrics::record_put_object_path(write_path.multipart_metric_label()); let encode_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now); - let (reader, w_size) = match write_path { - SmallWritePath::SingleBlockNonInline => { - Arc::new(erasure) - .encode_single_block_non_inline(stream, &mut writers, write_quorum) - .await? + let (reader, w_size) = async { + match write_path { + SmallWritePath::SingleBlockNonInline => { + Arc::new(erasure) + .encode_single_block_non_inline(stream, &mut writers, write_quorum) + .await + } + SmallWritePath::PipelineBatchedLarge => { + Arc::new(erasure).encode_batched(stream, &mut writers, write_quorum).await + } + SmallWritePath::Inline | SmallWritePath::Pipeline => { + Arc::new(erasure).encode(stream, &mut writers, write_quorum).await + } } - SmallWritePath::PipelineBatchedLarge => { - Arc::new(erasure).encode_batched(stream, &mut writers, write_quorum).await? - } - SmallWritePath::Inline | SmallWritePath::Pipeline => { - Arc::new(erasure).encode(stream, &mut writers, write_quorum).await? - } - }; + } + .instrument(tracing::info_span!( + "rustfs.ec.distribute", + event = "rustfs_ec_distribute", + component = "ecstore", + "rustfs.distribution.targets" = shuffle_disks.len(), + "rustfs.distribution.transport" = "erasure" + )) + .await?; if let Some(stage_start) = encode_stage_start { rustfs_io_metrics::record_put_object_stage_duration( @@ -576,6 +598,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { Ok(ret) } + .instrument(ec_write_span) .await; if result.is_err() @@ -1512,8 +1535,37 @@ mod tests { use crate::storage_api_contracts::namespace::NamespaceLocking as _; use rustfs_lock::{LockClient, client::local::LocalClient}; use serial_test::serial; + use std::sync::atomic::{AtomicU8, Ordering}; use tempfile::TempDir; use tokio::sync::RwLock; + use tracing_subscriber::{Layer, Registry, layer::SubscriberExt}; + + const EC_WRITE_SPAN_SEEN: u8 = 1; + const EC_DISTRIBUTE_SPAN_SEEN: u8 = 2; + + #[derive(Clone)] + struct EcWriteSpanLayer { + seen: Arc, + } + + impl Layer for EcWriteSpanLayer + where + S: tracing::Subscriber, + { + fn on_new_span( + &self, + attrs: &tracing::span::Attributes<'_>, + _: &tracing::Id, + _: tracing_subscriber::layer::Context<'_, S>, + ) { + let flag = match attrs.metadata().name() { + "rustfs.ec.write" => EC_WRITE_SPAN_SEEN, + "rustfs.ec.distribute" => EC_DISTRIBUTE_SPAN_SEEN, + _ => return, + }; + self.seen.fetch_or(flag, Ordering::Relaxed); + } + } async fn non_trash_tmp_entries(temp_dirs: &[TempDir]) -> Vec { let mut leftovers = Vec::new(); @@ -1742,6 +1794,9 @@ mod tests { #[tokio::test] async fn put_object_part_failure_cleans_tmp_workspace_inline() { + let seen = Arc::new(AtomicU8::new(0)); + let dispatch = tracing::Dispatch::new(Registry::default().with(EcWriteSpanLayer { seen: Arc::clone(&seen) })); + let _guard = tracing::dispatcher::set_default(&dispatch); let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let bucket = "multipart-tmp-clean-bucket"; let object = "object"; @@ -1765,6 +1820,12 @@ mod tests { .await .expect_err("short multipart stream should fail"); + assert_eq!( + seen.load(Ordering::Relaxed), + EC_WRITE_SPAN_SEEN | EC_DISTRIBUTE_SPAN_SEEN, + "multipart write must create EC write and distribution spans" + ); + let leftovers = non_trash_tmp_entries(&temp_dirs).await; assert!( leftovers.is_empty(), diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index d2f0333b9..bde8b81bc 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -23,6 +23,7 @@ use super::super::*; use crate::disk::OldCurrentSize; use crate::object_api::GetObjectBodySource; +use tracing::Instrument as _; /// Length of the full plaintext body when — and only when — this read's output /// is exactly the object's complete plaintext, so the app-layer body cache may @@ -740,6 +741,17 @@ impl SetDisks { let tmp_object = format!("{}/{}/part.1", tmp_dir, fi.data_dir.unwrap()); + let ec_write_span = tracing::info_span!( + "rustfs.ec.write", + event = "rustfs_ec_write", + component = "ecstore", + "rustfs.ec.data_shards" = data_drives, + "rustfs.ec.parity_shards" = parity_drives, + "rustfs.ec.write_quorum" = write_quorum, + "rustfs.ec.block_bytes" = fi.erasure.block_size, + "rustfs.distribution.targets" = shuffle_disks.len(), + "rustfs.distribution.transport" = "erasure" + ); let result: Result<(ObjectInfo, Option)> = async { let erasure = coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size); @@ -789,7 +801,15 @@ impl SetDisks { } }) .collect(); - let writer_results = join_all(writer_futs).await; + let writer_results = join_all(writer_futs) + .instrument(tracing::info_span!( + "rustfs.ec.distribute", + event = "rustfs_ec_distribute", + component = "ecstore", + "rustfs.distribution.targets" = shuffle_disks.len(), + "rustfs.distribution.transport" = "erasure" + )) + .await; let mut writers = Vec::with_capacity(writer_results.len()); let mut errors = Vec::with_capacity(writer_results.len()); for (w, e) in writer_results { @@ -1139,6 +1159,7 @@ impl SetDisks { old_current_size, )) } + .instrument(ec_write_span) .await; if issue3031_diag_enabled() diff --git a/crates/obs/src/lib.rs b/crates/obs/src/lib.rs index 1e2640c9a..1a6c78559 100644 --- a/crates/obs/src/lib.rs +++ b/crates/obs/src/lib.rs @@ -65,6 +65,7 @@ mod global; mod logging; pub mod metrics; pub mod semconv; +mod task; mod telemetry; pub use cleaner::*; @@ -79,7 +80,8 @@ pub use metrics::{ MetricsRuntimeShutdownHandle, MetricsRuntimeStatusSnapshot, MetricsRuntimeWorkerMutation, init_metrics_runtime, metrics_runtime_controller_snapshot, metrics_runtime_status_snapshot, }; -pub use semconv::{ErrorClass, Operation, ResultClass, Stage, observe_operation, stage_span}; +pub use semconv::{ErrorClass, Operation, ResultClass, Stage, StreamDirection, observe_operation, stage_span, stream_span}; +pub use task::spawn_traced; pub use telemetry::{OtelGuard, Recorder}; // Dial9 Tokio runtime telemetry diff --git a/crates/obs/src/semconv.rs b/crates/obs/src/semconv.rs index b44e08043..f82e8001d 100644 --- a/crates/obs/src/semconv.rs +++ b/crates/obs/src/semconv.rs @@ -6,6 +6,7 @@ use metrics::{Gauge, counter, gauge, histogram}; use opentelemetry::trace::Status; +use rustfs_utils::trace_attributes; use std::future::Future; use std::time::Instant; use tracing::Instrument; @@ -41,6 +42,7 @@ pub enum Operation { DeleteObject, CreateMultipartUpload, UploadPart, + UploadPartCopy, CompleteMultipartUpload, AbortMultipartUpload, } @@ -60,6 +62,7 @@ impl Operation { Self::DeleteObject => "delete_object", Self::CreateMultipartUpload => "create_multipart_upload", Self::UploadPart => "upload_part", + Self::UploadPartCopy => "upload_part_copy", Self::CompleteMultipartUpload => "complete_multipart_upload", Self::AbortMultipartUpload => "abort_multipart_upload", } @@ -79,6 +82,22 @@ pub enum Stage { ReplicationRemote, } +/// Bounded direction for a stream traversing the object pipeline. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum StreamDirection { + Ingress, + Copy, +} + +impl StreamDirection { + const fn as_str(self) -> &'static str { + match self { + Self::Ingress => trace_attributes::stream_direction::INGRESS, + Self::Copy => trace_attributes::stream_direction::COPY, + } + } +} + impl Stage { pub const fn as_str(self) -> &'static str { match self { @@ -146,7 +165,25 @@ impl ErrorClass { /// Create a child span for a bounded internal stage. pub fn stage_span(stage: Stage) -> tracing::Span { - tracing::info_span!("rustfs.stage", event = "rustfs_stage", component = "storage", stage = stage.as_str()) + tracing::info_span!( + "rustfs.stage", + event = "rustfs_stage", + component = "storage", + stage = stage.as_str(), + "rustfs.stage" = stage.as_str() + ) +} + +/// Create a child span for a request stream without exposing payload contents. +pub fn stream_span(direction: StreamDirection, expected_bytes: i64, buffer_bytes: usize) -> tracing::Span { + tracing::info_span!( + "rustfs.stream", + event = "rustfs_stream", + component = "storage", + "rustfs.stream.direction" = direction.as_str(), + "rustfs.stream.expected_bytes" = expected_bytes, + "rustfs.stream.buffer_bytes" = buffer_bytes + ) } /// Observe a complete S3 operation without exposing request data. @@ -159,7 +196,8 @@ where "rustfs.s3.operation", event = "s3_operation", component = "storage", - operation = operation_name + operation = operation_name, + "rustfs.operation" = operation_name ); let in_flight = gauge!(METRIC_REQUESTS_IN_FLIGHT, "operation" => operation_name); in_flight.increment(1.0); @@ -218,8 +256,10 @@ mod tests { #[test] fn operation_names_are_stable_and_bounded() { assert_eq!(Operation::PutObject.as_str(), "put_object"); + assert_eq!(Operation::UploadPartCopy.as_str(), "upload_part_copy"); assert_eq!(Operation::CompleteMultipartUpload.as_str(), "complete_multipart_upload"); assert_eq!(Stage::ReplicationRemote.as_str(), "replication_remote"); + assert_eq!(StreamDirection::Ingress.as_str(), trace_attributes::stream_direction::INGRESS); assert_eq!(ErrorClass::Quorum.as_str(), "quorum_failed"); } diff --git a/crates/obs/src/task.rs b/crates/obs/src/task.rs new file mode 100644 index 000000000..285155f2a --- /dev/null +++ b/crates/obs/src/task.rs @@ -0,0 +1,52 @@ +// 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. + +//! Task helpers for preserving tracing context across Tokio task boundaries. + +use tracing::Instrument; + +/// Spawn a Tokio task that inherits the current tracing span. +/// +/// Tokio does not propagate [`tracing::Span::current`] into spawned tasks. +/// Use this for work that remains part of the caller's operation. Detached +/// background side effects should start a new span instead. +pub fn spawn_traced(future: F) -> tokio::task::JoinHandle +where + F: std::future::Future + Send + 'static, + F::Output: Send + 'static, +{ + tokio::spawn(future.instrument(tracing::Span::current())) +} + +#[cfg(test)] +mod tests { + use super::spawn_traced; + + #[tokio::test] + async fn spawned_task_inherits_current_span() { + tracing::subscriber::set_global_default(tracing_subscriber::Registry::default()) + .expect("task tracing test should install its subscriber"); + + let (parent_id, task) = { + let parent = tracing::info_span!("parent"); + let parent_id = parent.id().expect("enabled parent span should have an id"); + let task = parent + .in_scope(|| spawn_traced(async { tracing::Span::current().id().expect("task should retain its parent span") })); + (parent_id, task) + }; + + let task_span_id = task.await.expect("task should complete"); + assert_eq!(task_span_id, parent_id); + } +} diff --git a/crates/obs/src/telemetry/filter.rs b/crates/obs/src/telemetry/filter.rs index d0e9c1df7..348a5c08b 100644 --- a/crates/obs/src/telemetry/filter.rs +++ b/crates/obs/src/telemetry/filter.rs @@ -228,6 +228,16 @@ pub(super) fn build_env_filter(logger_level: &str, default_level: Option<&str>) filter } +/// Build the filter used exclusively by the OTLP trace exporter. +/// +/// Trace collection must not depend on the log sink's severity: production +/// deployments commonly run with `RUST_LOG=warn` or `error` while still +/// expecting request spans to be exported. RustFS operation spans are emitted +/// at `info`, so retain that level regardless of `RUST_LOG`. +pub(super) fn build_trace_filter() -> EnvFilter { + EnvFilter::new("info") +} + #[cfg(test)] mod tests { use super::*; @@ -344,6 +354,13 @@ mod tests { }); } + #[test] + fn trace_filter_ignores_rust_log_severity() { + temp_env::with_var("RUST_LOG", Some("error"), || { + assert_eq!(build_trace_filter().to_string(), "info"); + }); + } + #[test] fn test_build_env_filter_target_only_rust_log_keeps_target_verbose() { // `RUST_LOG=hyper` is a target-only directive and should not be treated diff --git a/crates/obs/src/telemetry/otel.rs b/crates/obs/src/telemetry/otel.rs index f452a0c26..57d750532 100644 --- a/crates/obs/src/telemetry/otel.rs +++ b/crates/obs/src/telemetry/otel.rs @@ -39,7 +39,7 @@ use crate::cleaner::types::FileMatchMode; use crate::config::OtelConfig; use crate::global::set_observability_metric_enabled; -use crate::telemetry::filter::build_env_filter; +use crate::telemetry::filter::{build_env_filter, build_trace_filter}; use crate::telemetry::guard::{OtelGuard, ProfilingAgent}; use crate::telemetry::local::{build_json_log_layer, spawn_cleanup_task}; use crate::telemetry::recorder::{Recorder, install_process_global_recorder}; @@ -241,8 +241,10 @@ pub(super) fn init_observability_http( // ── Tracing subscriber registry ─────────────────────────────────────────── let tracer_layer = tracer_provider .as_ref() - .map(|p| OpenTelemetryLayer::new(p.tracer(service_name.to_string()))); - let metrics_layer = meter_provider.as_ref().map(|p| MetricsLayer::new(p.clone())); + .map(|p| OpenTelemetryLayer::new(p.tracer(service_name.to_string())).with_filter(build_trace_filter())); + let metrics_layer = meter_provider + .as_ref() + .map(|p| MetricsLayer::new(p.clone()).with_filter(build_env_filter(logger_level, None))); // Optional stdout mirror (matching init_file_logging_internal logic) // This is separate from OTLP stdout logic. If file logging is enabled, we honor its stdout rules. @@ -253,12 +255,11 @@ pub(super) fn init_observability_http( } let local_file_fallback_enabled = file_layer_opt.is_some(); let stdout_mirror_enabled = stdout_guard.is_some(); - let filter = build_env_filter(logger_level, None); tracing_subscriber::registry() - .with(filter) + .with(build_trace_filter()) .with(ErrorLayer::default()) - .with(file_layer_opt) - .with(stdout_layer_opt) + .with(file_layer_opt.map(|layer| layer.with_filter(build_env_filter(logger_level, None)))) + .with(stdout_layer_opt.map(|layer| layer.with_filter(build_env_filter(logger_level, None)))) .with(tracer_layer) .with(otel_bridge) .with(metrics_layer) diff --git a/crates/scanner/Cargo.toml b/crates/scanner/Cargo.toml index db5a510f4..4a8794273 100644 --- a/crates/scanner/Cargo.toml +++ b/crates/scanner/Cargo.toml @@ -52,6 +52,7 @@ rand = { workspace = true } s3s = { workspace = true } metrics = { workspace = true } rustfs-data-usage = { workspace = true } +rustfs-obs = { workspace = true } [dev-dependencies] tracing-subscriber = { workspace = true } diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 9f4811689..f13204c4a 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -786,7 +786,7 @@ impl ScannerIO for ECStore { let (tx, mut rx) = mpsc::channel::(1); // Spawn task to receive and store results - let receiver_fut = tokio::spawn(async move { + let receiver_fut = rustfs_obs::spawn_traced(async move { while let Some(result) = rx.recv().await { let mut results = results_mutex_clone.lock().await; results[results_index_clone] = result; @@ -797,7 +797,7 @@ impl ScannerIO for ECStore { let scan_plan = ScannerBucketScanPlan::new(all_buckets.clone(), dirty_usage_buckets.clone(), failed_dirty_buckets.clone()); // Spawn task to run the scanner - let scanner_fut = tokio::spawn(async move { + let scanner_fut = rustfs_obs::spawn_traced(async move { let permit_wait = child_token_clone.clone(); let permit_wait_start = Instant::now(); let _permit = tokio::select! { @@ -871,7 +871,7 @@ impl ScannerIO for ECStore { let budget_for_updates = budget.clone(); let child_token_for_updates = child_token.clone(); let dirty_usage_buckets_for_updates = dirty_usage_buckets.clone(); - tokio::spawn(async move { + rustfs_obs::spawn_traced(async move { let mut last_update = SystemTime::UNIX_EPOCH; let mut has_sent_once = false; @@ -1089,7 +1089,7 @@ impl ScannerIOCache for SetDisks { let ctx_clone = ctx.clone(); let completed_bucket_count = Arc::new(AtomicUsize::new(0)); let completed_bucket_count_clone = completed_bucket_count.clone(); - let collect_bucket_results_fut = tokio::spawn(async move { + let collect_bucket_results_fut = rustfs_obs::spawn_traced(async move { let mut cancelled = false; loop { @@ -1127,7 +1127,7 @@ impl ScannerIOCache for SetDisks { let pool_label_clone = pool_label.clone(); let set_label_clone = set_label.clone(); let failed_dirty_buckets_clone = failed_dirty_buckets.clone(); - futs.push(tokio::spawn(async move { + futs.push(rustfs_obs::spawn_traced(async move { loop { let Some(bucket) = bucket_rx_mutex_clone.lock().await.recv().await else { break; diff --git a/crates/utils/src/lib.rs b/crates/utils/src/lib.rs index 2ac6630fd..83fec449f 100644 --- a/crates/utils/src/lib.rs +++ b/crates/utils/src/lib.rs @@ -52,6 +52,7 @@ pub mod crypto; pub mod compress; pub mod logging; +pub mod trace_attributes; #[cfg(feature = "path")] pub mod dirs; diff --git a/crates/utils/src/trace_attributes.rs b/crates/utils/src/trace_attributes.rs new file mode 100644 index 000000000..893b2600a --- /dev/null +++ b/crates/utils/src/trace_attributes.rs @@ -0,0 +1,66 @@ +//! RustFS-specific OpenTelemetry attribute names and bounded values. +//! +//! These constants are for traces only. Do not reuse request identifiers, +//! bucket names, object keys, upload IDs, credentials, headers, or peer +//! addresses as metric labels. + +/// S3 operation name, for example `put_object`. +pub const OPERATION: &str = "rustfs.operation"; +/// Internal pipeline stage name. +pub const STAGE: &str = "rustfs.stage"; +/// Direction of a streaming pipeline. +pub const STREAM_DIRECTION: &str = "rustfs.stream.direction"; +/// Declared or resolved stream size in bytes. +pub const STREAM_EXPECTED_BYTES: &str = "rustfs.stream.expected_bytes"; +/// Buffer capacity selected for the stream. +pub const STREAM_BUFFER_BYTES: &str = "rustfs.stream.buffer_bytes"; +/// Number of erasure data shards. +pub const EC_DATA_SHARDS: &str = "rustfs.ec.data_shards"; +/// Number of erasure parity shards. +pub const EC_PARITY_SHARDS: &str = "rustfs.ec.parity_shards"; +/// Required successful erasure writes. +pub const EC_WRITE_QUORUM: &str = "rustfs.ec.write_quorum"; +/// Erasure block size in bytes. +pub const EC_BLOCK_BYTES: &str = "rustfs.ec.block_bytes"; +/// Number of shard targets receiving a distributed write. +pub const DISTRIBUTION_TARGETS: &str = "rustfs.distribution.targets"; +/// Bounded transport category for distributed work. +pub const DISTRIBUTION_TRANSPORT: &str = "rustfs.distribution.transport"; + +/// Bounded stream direction values. +pub mod stream_direction { + /// Client request body flowing into storage. + pub const INGRESS: &str = "ingress"; + /// Source-object body flowing through UploadPartCopy. + pub const COPY: &str = "copy"; +} + +/// Bounded distributed transport values. +pub mod distribution_transport { + /// Erasure shard writer setup and dispatch. + pub const ERASURE: &str = "erasure"; +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn attribute_names_use_the_rustfs_namespace() { + for name in [ + OPERATION, + STAGE, + STREAM_DIRECTION, + STREAM_EXPECTED_BYTES, + STREAM_BUFFER_BYTES, + EC_DATA_SHARDS, + EC_PARITY_SHARDS, + EC_WRITE_QUORUM, + EC_BLOCK_BYTES, + DISTRIBUTION_TARGETS, + DISTRIBUTION_TRANSPORT, + ] { + assert!(name.starts_with("rustfs.")); + } + } +} diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index ea737b5b3..113383fb1 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -338,6 +338,7 @@ impl DefaultMultipartUsecase { // abort operations to avoid leaking information about upload_id format requirements. match store .abort_multipart_upload(bucket.as_str(), key.as_str(), upload_id.as_str(), opts) + .instrument(rustfs_obs::stage_span(rustfs_obs::Stage::EcWrite)) .await { Ok(_) => { @@ -382,7 +383,11 @@ impl DefaultMultipartUsecase { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; - match store.get_object_info(&bucket, &key, &ObjectOptions::default()).await { + match store + .get_object_info(&bucket, &key, &ObjectOptions::default()) + .instrument(rustfs_obs::stage_span(rustfs_obs::Stage::Metadata)) + .await + { Ok(info) => { if !info.delete_marker { if let Some(ifmatch) = if_match @@ -448,7 +453,11 @@ impl DefaultMultipartUsecase { .await .map_err(ApiError::from)?, ); - let previous_current_size = match store.get_object_info(&bucket, &key, ¤t_opts).await { + let previous_current_size = match store + .get_object_info(&bucket, &key, ¤t_opts) + .instrument(rustfs_obs::stage_span(rustfs_obs::Stage::Metadata)) + .await + { Ok(existing_obj_info) => { validate_existing_object_lock_for_write(&existing_obj_info, ¤t_opts)?; Some(existing_obj_info.size.max(0) as u64) @@ -463,6 +472,7 @@ impl DefaultMultipartUsecase { let multipart_info = store .get_multipart_info(&bucket, &key, &upload_id, &ObjectOptions::default()) + .instrument(rustfs_obs::stage_span(rustfs_obs::Stage::Metadata)) .await .map_err(ApiError::from)?; let cache_adapter = self.object_data_cache(); @@ -756,6 +766,7 @@ impl DefaultMultipartUsecase { checksum_type, } = store .new_multipart_upload(&bucket, &key, &opts) + .instrument(rustfs_obs::stage_span(rustfs_obs::Stage::Metadata)) .await .map_err(ApiError::from)?; @@ -827,6 +838,7 @@ impl DefaultMultipartUsecase { let mut opts = ObjectOptions::default(); let fi = store .get_multipart_info(&bucket, &key, &upload_id, &opts) + .instrument(rustfs_obs::stage_span(rustfs_obs::Stage::Metadata)) .await .map_err(ApiError::from)?; @@ -969,6 +981,7 @@ impl DefaultMultipartUsecase { let info = store .put_object_part(&bucket, &key, &upload_id, part_id, &mut reader, &opts) .instrument(rustfs_obs::stage_span(rustfs_obs::Stage::EcWrite)) + .instrument(rustfs_obs::stream_span(rustfs_obs::StreamDirection::Ingress, actual_size, buffer_size)) .await .map_err(ApiError::from)?; @@ -1134,6 +1147,7 @@ impl DefaultMultipartUsecase { let mp_info = store .get_multipart_info(&bucket, &key, &upload_id, &ObjectOptions::default()) + .instrument(rustfs_obs::stage_span(rustfs_obs::Stage::Metadata)) .await .map_err(ApiError::from)?; @@ -1154,6 +1168,7 @@ impl DefaultMultipartUsecase { let src_reader = store .get_object_reader(&src_bucket, &src_key, rs.clone(), h, &get_opts) + .instrument(rustfs_obs::stage_span(rustfs_obs::Stage::EcRead)) .await .map_err(map_get_object_reader_error)?; @@ -1210,6 +1225,7 @@ impl DefaultMultipartUsecase { let src_reader = store .get_object_reader(&src_bucket, &src_key, rs.clone(), h, &get_opts) + .instrument(rustfs_obs::stage_span(rustfs_obs::Stage::EcRead)) .await .map_err(map_get_object_reader_error)?; let src_stream = src_reader.stream; @@ -1335,7 +1351,8 @@ impl DefaultMultipartUsecase { let part_info = store .put_object_part(&bucket, &key, &upload_id, part_id, &mut reader, &dst_opts) - .instrument(rustfs_obs::stage_span(rustfs_obs::Stage::EcRead)) + .instrument(rustfs_obs::stage_span(rustfs_obs::Stage::EcWrite)) + .instrument(rustfs_obs::stream_span(rustfs_obs::StreamDirection::Copy, actual_size, 0)) .await .map_err(ApiError::from)?; diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 45306f2a6..98ee12c67 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -3676,7 +3676,10 @@ impl DefaultObjectUsecase { ); let previous_current_info = { crate::hp_guard!("S3::put_object_prelookup"); - store.get_object_info(&bucket, &key, ¤t_opts).await + store + .get_object_info(&bucket, &key, ¤t_opts) + .instrument(rustfs_obs::stage_span(rustfs_obs::Stage::Metadata)) + .await }; Some(match previous_current_info { Ok(existing_obj_info) => { @@ -3877,6 +3880,8 @@ impl DefaultObjectUsecase { let (obj_info, backfilled_old_current_size) = match store .put_object_with_old_current_size(&bucket, &key, &mut reader, &opts) + .instrument(rustfs_obs::stage_span(rustfs_obs::Stage::EcWrite)) + .instrument(rustfs_obs::stream_span(rustfs_obs::StreamDirection::Ingress, actual_size, buffer_size)) .await .map_err(ApiError::from) { diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index e84df936e..386359b66 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -1589,6 +1589,6 @@ impl S3 for FS { async fn upload_part_copy(&self, req: S3Request) -> S3Result> { record_s3_op(S3Operation::UploadPartCopy); let usecase = s3_api::multipart_usecase_for(self); - Box::pin(usecase.execute_upload_part_copy(req)).await + observe_operation(ObsOperation::UploadPartCopy, Box::pin(usecase.execute_upload_part_copy(req))).await } } diff --git a/rustfs/src/storage/ecfs_test.rs b/rustfs/src/storage/ecfs_test.rs index 4541a7d02..ef8c625db 100644 --- a/rustfs/src/storage/ecfs_test.rs +++ b/rustfs/src/storage/ecfs_test.rs @@ -35,15 +35,21 @@ mod tests { }; use rustfs_zip::CompressionFormat; use s3s::dto::{ - CORSConfiguration, CORSRule, DefaultRetention, DeleteObjectTaggingInput, Delimiter, FilterRule, FilterRuleName, - GetBucketAclInput, GetObjectAclInput, GetObjectLegalHoldInput, GetObjectRetentionInput, GetObjectTaggingInput, - LambdaFunctionConfiguration, NotificationConfigurationFilter, ObjectLockConfiguration, ObjectLockEnabled, - ObjectLockLegalHold, ObjectLockLegalHoldStatus, ObjectLockRetention, ObjectLockRetentionMode, ObjectLockRule, - PutBucketAclInput, PutObjectAclInput, PutObjectLegalHoldInput, PutObjectLockConfigurationInput, PutObjectRetentionInput, - PutObjectTaggingInput, QueueConfiguration, S3KeyFilter, Tag, Tagging, TopicConfiguration, + CORSConfiguration, CORSRule, CopySource, DefaultRetention, DeleteObjectTaggingInput, Delimiter, FilterRule, + FilterRuleName, GetBucketAclInput, GetObjectAclInput, GetObjectLegalHoldInput, GetObjectRetentionInput, + GetObjectTaggingInput, LambdaFunctionConfiguration, NotificationConfigurationFilter, ObjectLockConfiguration, + ObjectLockEnabled, ObjectLockLegalHold, ObjectLockLegalHoldStatus, ObjectLockRetention, ObjectLockRetentionMode, + ObjectLockRule, PutBucketAclInput, PutObjectAclInput, PutObjectLegalHoldInput, PutObjectLockConfigurationInput, + PutObjectRetentionInput, PutObjectTaggingInput, QueueConfiguration, S3KeyFilter, Tag, Tagging, TopicConfiguration, + UploadPartCopyInput, }; use s3s::{S3, S3Error, S3ErrorCode, S3Request, s3_error}; + use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; + use tracing_subscriber::{Layer, Registry, layer::SubscriberExt}; fn build_request(input: T, method: Method) -> S3Request { S3Request { @@ -59,6 +65,27 @@ mod tests { } } + #[derive(Clone)] + struct OperationSpanLayer { + seen: Arc, + } + + impl Layer for OperationSpanLayer + where + S: tracing::Subscriber, + { + fn on_new_span( + &self, + attrs: &tracing::span::Attributes<'_>, + _: &tracing::Id, + _: tracing_subscriber::layer::Context<'_, S>, + ) { + if attrs.metadata().name() == "rustfs.s3.operation" { + self.seen.store(true, Ordering::Relaxed); + } + } + } + #[test] fn test_fs_creation() { let _fs = FS::new(); @@ -88,6 +115,30 @@ mod tests { assert_eq!(format!("{fs:?}"), format!("{cloned_fs:?}")); } + #[tokio::test] + async fn upload_part_copy_starts_an_operation_span() { + let seen = Arc::new(AtomicBool::new(false)); + let subscriber = Registry::default().with(OperationSpanLayer { seen: seen.clone() }); + let _subscriber_guard = tracing::subscriber::set_default(subscriber); + let fs = FS::new(); + let input = UploadPartCopyInput::builder() + .bucket("bucket".to_string()) + .key("object".to_string()) + .copy_source(CopySource::Bucket { + bucket: "src-bucket".into(), + key: "src-object".into(), + version_id: None, + }) + .part_number(1) + .upload_id("upload-id".to_string()) + .build() + .expect("valid UploadPartCopy request"); + + let _ = fs.upload_part_copy(build_request(input, Method::PUT)).await; + + assert!(seen.load(Ordering::Relaxed)); + } + #[test] fn test_rustfs_owner_helpers_are_stable() { let owner = rustfs_owner(); diff --git a/rustfs/src/storage/request_context.rs b/rustfs/src/storage/request_context.rs index 6f3965140..208674a05 100644 --- a/rustfs/src/storage/request_context.rs +++ b/rustfs/src/storage/request_context.rs @@ -188,7 +188,7 @@ pub fn spawn_traced(fut: F) where F: std::future::Future + Send + 'static, { - tokio::spawn(tracing::Instrument::instrument(fut, tracing::Span::current())); + rustfs_obs::spawn_traced(fut); } #[cfg(test)]