feat(obs): enhance observability with tracing spans and metrics integration

.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'",

Cargo.lock

@@ -9592,6 +9592,7 @@ dependencies = [
  "jiff",
  "libc",
  "metrics",
+ "metrics-util",
  "num_cpus",
  "nvml-wrapper",
  "opentelemetry",
@@ -9884,6 +9885,7 @@ dependencies = [
  "rustfs-data-usage",
  "rustfs-ecstore",
  "rustfs-filemeta",
+ "rustfs-obs",
  "rustfs-storage-api",
  "rustfs-utils",
  "s3s",

crates/config/src/observability/mod.rs

@@ -37,6 +37,10 @@ pub const ENV_OBS_METER_INTERVAL: &str = "RUSTFS_OBS_METER_INTERVAL";
 pub const ENV_OBS_SERVICE_NAME: &str = "RUSTFS_OBS_SERVICE_NAME";
 pub const ENV_OBS_SERVICE_VERSION: &str = "RUSTFS_OBS_SERVICE_VERSION";
 pub const ENV_OBS_ENVIRONMENT: &str = "RUSTFS_OBS_ENVIRONMENT";
+/// Stable OpenTelemetry service instance identity for this RustFS process.
+pub const ENV_OBS_INSTANCE_ID: &str = "RUSTFS_OBS_INSTANCE_ID";
+/// Optional stable RustFS deployment identity used to correlate cluster telemetry.
+pub const ENV_OBS_CLUSTER_ID: &str = "RUSTFS_OBS_CLUSTER_ID";
 
 /// Per-signal enable/disable flags (default: true)
 pub const ENV_OBS_TRACES_EXPORT_ENABLED: &str = "RUSTFS_OBS_TRACES_EXPORT_ENABLED";
@@ -131,6 +135,8 @@ mod tests {
         assert_eq!(ENV_OBS_SERVICE_NAME, "RUSTFS_OBS_SERVICE_NAME");
         assert_eq!(ENV_OBS_SERVICE_VERSION, "RUSTFS_OBS_SERVICE_VERSION");
         assert_eq!(ENV_OBS_ENVIRONMENT, "RUSTFS_OBS_ENVIRONMENT");
+        assert_eq!(ENV_OBS_INSTANCE_ID, "RUSTFS_OBS_INSTANCE_ID");
+        assert_eq!(ENV_OBS_CLUSTER_ID, "RUSTFS_OBS_CLUSTER_ID");
         assert_eq!(ENV_OBS_LOGGER_LEVEL, "RUSTFS_OBS_LOGGER_LEVEL");
         assert_eq!(ENV_OBS_LOG_STDOUT_ENABLED, "RUSTFS_OBS_LOG_STDOUT_ENABLED");
         assert_eq!(ENV_OBS_LOG_DIRECTORY, "RUSTFS_OBS_LOG_DIRECTORY");

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<PartInfo> = 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 {
+            let (reader, w_size) = async {
-                SmallWritePath::SingleBlockNonInline => {
+                match write_path {
-                    Arc::new(erasure)
+                    SmallWritePath::SingleBlockNonInline => {
-                        .encode_single_block_non_inline(stream, &mut writers, write_quorum)
+                        Arc::new(erasure)
-                        .await?
+                            .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?
+            .instrument(tracing::info_span!(
-                }
+                "rustfs.ec.distribute",
-                SmallWritePath::Inline | SmallWritePath::Pipeline => {
+                event = "rustfs_ec_distribute",
-                    Arc::new(erasure).encode(stream, &mut writers, write_quorum).await?
+                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<AtomicU8>,
+    }
+
+    impl<S> Layer<S> 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<String> {
         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(),

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<OldCurrentSize>)> = 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()

crates/obs/Cargo.toml

@@ -115,6 +115,7 @@ libc = { workspace = true }
 
 
 [dev-dependencies]
+metrics-util = { version = "0.20", features = ["debugging"] }
 tokio = { workspace = true, features = ["full"] }
 tempfile = { workspace = true }
 temp-env = { workspace = true }

crates/obs/README.md

@@ -121,7 +121,7 @@ All configuration is read from environment variables at startup.
 | `RUSTFS_OBS_LOGS_EXPORT_ENABLED`      | `true`    | Toggle OTLP log export                                     |
 | `RUSTFS_OBS_PROFILING_EXPORT_ENABLED` | `false`   | Toggle profiling export                                    |
 | `RUSTFS_OBS_USE_STDOUT`               | `false`   | Mirror all signals to stdout alongside OTLP                |
-| `RUSTFS_OBS_SAMPLE_RATIO`             | `0.1`     | Trace sampling ratio `0.0`–`1.0`                           |
+| `RUSTFS_OBS_SAMPLE_RATIO`             | `1.0`     | Trace sampling ratio `0.0`–`1.0`                           |
 | `RUSTFS_OBS_METER_INTERVAL`           | `15`      | Metrics export interval (seconds)                          |
 
 ### Service identity
@@ -131,12 +131,14 @@ All configuration is read from environment variables at startup.
 | `RUSTFS_OBS_SERVICE_NAME`    | `rustfs`          | OTel `service.name`                                     |
 | `RUSTFS_OBS_SERVICE_VERSION` | _(crate version)_ | OTel `service.version`                                  |
 | `RUSTFS_OBS_ENVIRONMENT`     | `development`     | Deployment environment (`production`, `development`, …) |
+| `RUSTFS_OBS_INSTANCE_ID`     | _(empty)_         | Stable OTel `service.instance.id`; omitted when unset    |
+| `RUSTFS_OBS_CLUSTER_ID`      | _(empty)_         | Stable RustFS deployment identity; omitted when unset    |
 
 ### Local logging
 
 | Variable                        | Default      | Description                                                                                                                                                                                                                                                                                                                                                           |
 |---------------------------------|--------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| `RUSTFS_OBS_LOGGER_LEVEL`       | `info`       | Log level; `RUST_LOG` syntax supported                                                                                                                                                                                                                                                                                                                                |
+| `RUSTFS_OBS_LOGGER_LEVEL`       | `error`      | Log level; `RUST_LOG` syntax supported                                                                                                                                                                                                                                                                                                                               |
 | `RUSTFS_OBS_LOG_STDOUT_ENABLED` | `false`      | When file logging is active, also mirror to stdout                                                                                                                                                                                                                                                                                                                    |
 | `RUSTFS_OBS_LOG_DIRECTORY`      | _(empty)_    | **Directory for rolling log files. When empty, logs go to stdout only**                                                                                                                                                                                                                                                                                               |
 | `RUSTFS_OBS_LOG_FILENAME`       | `rustfs.log` | Base filename for rolling logs. Rotated archives include a high-precision timestamp and counter. With the default `RUSTFS_OBS_LOG_MATCH_MODE=suffix`, names look like `<timestamp>-<counter>.rustfs.log` (e.g., `20231027103001.123456-0.rustfs.log`); with `prefix`, they look like `rustfs.log.<timestamp>-<counter>` (e.g., `rustfs.log.20231027103001.123456-0`). |
@@ -167,6 +169,22 @@ All configuration is read from environment variables at startup.
 
 ---
 
+## Local all-in-one
+
+After starting Grafana/Tempo/Loki/Prometheus from `/Users/tang/code/otel-all-in-one`, configure RustFS to export all three OTLP signals directly:
+
+```bash
+export RUSTFS_OBS_ENDPOINT=http://127.0.0.1:4318
+export RUSTFS_OBS_SERVICE_NAME=rustfs
+export RUSTFS_OBS_ENVIRONMENT=local
+export RUSTFS_OBS_SAMPLE_RATIO=1
+export RUSTFS_OBS_LOGGER_LEVEL=info
+```
+
+Do not also configure promtail to collect the same RustFS logs: OTLP logs can already be correlated from Tempo to Loki. When `RUSTFS_OBS_ENDPOINT` is unset, RustFS retains its existing local logging fallback.
+
+---
+
 ## Cleaner & Rotation Metrics
 
 The log rotation and cleanup pipeline emits these metrics (via the `metrics` facade):

crates/obs/src/config.rs

@@ -28,14 +28,14 @@ use rustfs_config::observability::{
     DEFAULT_OBS_LOG_DRY_RUN, DEFAULT_OBS_LOG_GZIP_COMPRESSION_LEVEL, DEFAULT_OBS_LOG_MATCH_MODE,
     DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES, DEFAULT_OBS_LOG_MAX_TOTAL_SIZE_BYTES, DEFAULT_OBS_LOG_MIN_FILE_AGE_SECONDS,
     DEFAULT_OBS_LOG_PARALLEL_COMPRESS, DEFAULT_OBS_LOG_PARALLEL_WORKERS, DEFAULT_OBS_LOG_ZSTD_COMPRESSION_LEVEL,
-    DEFAULT_OBS_LOG_ZSTD_FALLBACK_TO_GZIP, DEFAULT_OBS_LOG_ZSTD_WORKERS, ENV_OBS_ENDPOINT, ENV_OBS_ENDPOINT_HEADERS,
+    DEFAULT_OBS_LOG_ZSTD_FALLBACK_TO_GZIP, DEFAULT_OBS_LOG_ZSTD_WORKERS, ENV_OBS_CLUSTER_ID, ENV_OBS_ENDPOINT,
-    ENV_OBS_ENDPOINT_LOGS_HEADERS, ENV_OBS_ENDPOINT_LOGS_TIMEOUT_MILLIS, ENV_OBS_ENDPOINT_METRICS_HEADERS,
+    ENV_OBS_ENDPOINT_HEADERS, ENV_OBS_ENDPOINT_LOGS_HEADERS, ENV_OBS_ENDPOINT_LOGS_TIMEOUT_MILLIS,
-    ENV_OBS_ENDPOINT_METRICS_TIMEOUT_MILLIS, ENV_OBS_ENDPOINT_TIMEOUT_MILLIS, ENV_OBS_ENDPOINT_TRACES_HEADERS,
+    ENV_OBS_ENDPOINT_METRICS_HEADERS, ENV_OBS_ENDPOINT_METRICS_TIMEOUT_MILLIS, ENV_OBS_ENDPOINT_TIMEOUT_MILLIS,
-    ENV_OBS_ENDPOINT_TRACES_TIMEOUT_MILLIS, ENV_OBS_ENVIRONMENT, ENV_OBS_LOG_CLEANUP_INTERVAL_SECONDS,
+    ENV_OBS_ENDPOINT_TRACES_HEADERS, ENV_OBS_ENDPOINT_TRACES_TIMEOUT_MILLIS, ENV_OBS_ENVIRONMENT, ENV_OBS_INSTANCE_ID,
-    ENV_OBS_LOG_COMPRESS_OLD_FILES, ENV_OBS_LOG_COMPRESSED_FILE_RETENTION_DAYS, ENV_OBS_LOG_COMPRESSION_ALGORITHM,
+    ENV_OBS_LOG_CLEANUP_INTERVAL_SECONDS, ENV_OBS_LOG_COMPRESS_OLD_FILES, ENV_OBS_LOG_COMPRESSED_FILE_RETENTION_DAYS,
-    ENV_OBS_LOG_DELETE_EMPTY_FILES, ENV_OBS_LOG_DIRECTORY, ENV_OBS_LOG_DRY_RUN, ENV_OBS_LOG_ENDPOINT,
+    ENV_OBS_LOG_COMPRESSION_ALGORITHM, ENV_OBS_LOG_DELETE_EMPTY_FILES, ENV_OBS_LOG_DIRECTORY, ENV_OBS_LOG_DRY_RUN,
-    ENV_OBS_LOG_EXCLUDE_PATTERNS, ENV_OBS_LOG_FILENAME, ENV_OBS_LOG_GZIP_COMPRESSION_LEVEL, ENV_OBS_LOG_KEEP_FILES,
+    ENV_OBS_LOG_ENDPOINT, ENV_OBS_LOG_EXCLUDE_PATTERNS, ENV_OBS_LOG_FILENAME, ENV_OBS_LOG_GZIP_COMPRESSION_LEVEL,
-    ENV_OBS_LOG_MATCH_MODE, ENV_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES, ENV_OBS_LOG_MAX_TOTAL_SIZE_BYTES,
+    ENV_OBS_LOG_KEEP_FILES, ENV_OBS_LOG_MATCH_MODE, ENV_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES, ENV_OBS_LOG_MAX_TOTAL_SIZE_BYTES,
     ENV_OBS_LOG_MIN_FILE_AGE_SECONDS, ENV_OBS_LOG_PARALLEL_COMPRESS, ENV_OBS_LOG_PARALLEL_WORKERS, ENV_OBS_LOG_ROTATION_TIME,
     ENV_OBS_LOG_STDOUT_ENABLED, ENV_OBS_LOG_ZSTD_COMPRESSION_LEVEL, ENV_OBS_LOG_ZSTD_FALLBACK_TO_GZIP, ENV_OBS_LOG_ZSTD_WORKERS,
     ENV_OBS_LOGGER_LEVEL, ENV_OBS_LOGS_EXPORT_ENABLED, ENV_OBS_METER_INTERVAL, ENV_OBS_METRIC_ENDPOINT,
@@ -162,6 +162,10 @@ pub struct OtelConfig {
     pub service_version: Option<String>,
     /// Deployment environment tag, e.g. `production` or `development`.
     pub environment: Option<String>,
+    /// Stable identity for this process instance. Empty values are omitted.
+    pub instance_id: Option<String>,
+    /// Stable RustFS deployment identity. Empty values are omitted.
+    pub cluster_id: Option<String>,
 
     // ── Local logging ─────────────────────────────────────────────────────────
     /// Minimum log level directive (default: `error`).
@@ -307,6 +311,8 @@ impl OtelConfig {
             service_name: Some(get_env_str(ENV_OBS_SERVICE_NAME, APP_NAME)),
             service_version: Some(get_env_str(ENV_OBS_SERVICE_VERSION, SERVICE_VERSION)),
             environment: Some(get_env_str(ENV_OBS_ENVIRONMENT, ENVIRONMENT)),
+            instance_id: get_env_opt_str(ENV_OBS_INSTANCE_ID),
+            cluster_id: get_env_opt_str(ENV_OBS_CLUSTER_ID),
             // Local logging
             logger_level: Some(get_env_str(ENV_OBS_LOGGER_LEVEL, DEFAULT_LOG_LEVEL)),
             log_stdout_enabled: get_env_opt_bool(ENV_OBS_LOG_STDOUT_ENABLED),

crates/obs/src/lib.rs

@@ -64,6 +64,8 @@ mod error;
 mod global;
 mod logging;
 pub mod metrics;
+pub mod semconv;
+mod task;
 mod telemetry;
 
 pub use cleaner::*;
@@ -78,6 +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, StreamDirection, observe_operation, stage_span, stream_span};
+pub use task::spawn_traced;
 pub use telemetry::{OtelGuard, Recorder};
 
 // Dial9 Tokio runtime telemetry

crates/obs/src/semconv.rs

@@ -0,0 +1,305 @@
+//! Stable, low-cardinality observability vocabulary for RustFS request paths.
+//!
+//! This module intentionally accepts no bucket, object, request, URL, or
+//! credential values. Those values are either unbounded or sensitive and must
+//! not become telemetry dimensions.
+
+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;
+use tracing_opentelemetry::OpenTelemetrySpanExt;
+
+const METRIC_REQUESTS_TOTAL: &str = "rustfs_s3_requests_total";
+const METRIC_REQUEST_FAILURES_TOTAL: &str = "rustfs_s3_request_failures_total";
+const METRIC_REQUEST_DURATION_SECONDS: &str = "rustfs_s3_request_duration_seconds";
+const METRIC_REQUESTS_IN_FLIGHT: &str = "rustfs_s3_requests_in_flight";
+
+struct InFlightRequest {
+    gauge: Gauge,
+}
+
+impl Drop for InFlightRequest {
+    fn drop(&mut self) {
+        self.gauge.decrement(1.0);
+    }
+}
+
+/// A stable S3 operation name suitable for telemetry dimensions.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum Operation {
+    CreateBucket,
+    DeleteBucket,
+    HeadBucket,
+    ListBuckets,
+    ListObjects,
+    ListObjectsV2,
+    PutObject,
+    GetObject,
+    HeadObject,
+    DeleteObject,
+    CreateMultipartUpload,
+    UploadPart,
+    UploadPartCopy,
+    CompleteMultipartUpload,
+    AbortMultipartUpload,
+}
+
+impl Operation {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::CreateBucket => "create_bucket",
+            Self::DeleteBucket => "delete_bucket",
+            Self::HeadBucket => "head_bucket",
+            Self::ListBuckets => "list_buckets",
+            Self::ListObjects => "list_objects",
+            Self::ListObjectsV2 => "list_objects_v2",
+            Self::PutObject => "put_object",
+            Self::GetObject => "get_object",
+            Self::HeadObject => "head_object",
+            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",
+        }
+    }
+}
+
+/// A bounded internal operation stage. Do not create spans per shard or block.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum Stage {
+    Authorize,
+    Metadata,
+    EcRead,
+    EcWrite,
+    Quorum,
+    Response,
+    ReplicationDispatch,
+    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 {
+            Self::Authorize => "authorize",
+            Self::Metadata => "metadata",
+            Self::EcRead => "ec_read",
+            Self::EcWrite => "ec_write",
+            Self::Quorum => "quorum",
+            Self::Response => "response",
+            Self::ReplicationDispatch => "replication_dispatch",
+            Self::ReplicationRemote => "replication_remote",
+        }
+    }
+}
+
+/// Bounded result classification shared by metrics and spans.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ResultClass {
+    Success,
+    Error,
+}
+
+impl ResultClass {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Success => "success",
+            Self::Error => "error",
+        }
+    }
+}
+
+/// Stable error categories for future fine-grained operation instrumentation.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ErrorClass {
+    Validation,
+    NotFound,
+    Conflict,
+    Timeout,
+    Canceled,
+    ResourceExhausted,
+    Io,
+    Network,
+    Quorum,
+    Consistency,
+    Internal,
+}
+
+impl ErrorClass {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Validation => "validation_error",
+            Self::NotFound => "not_found",
+            Self::Conflict => "conflict",
+            Self::Timeout => "timeout",
+            Self::Canceled => "canceled",
+            Self::ResourceExhausted => "resource_exhausted",
+            Self::Io => "io_error",
+            Self::Network => "network_error",
+            Self::Quorum => "quorum_failed",
+            Self::Consistency => "consistency_violation",
+            Self::Internal => "internal_error",
+        }
+    }
+}
+
+/// 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(),
+        "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.
+pub async fn observe_operation<T, E, F>(operation: Operation, future: F) -> Result<T, E>
+where
+    F: Future<Output = Result<T, E>>,
+{
+    let operation_name = operation.as_str();
+    let span = tracing::info_span!(
+        "rustfs.s3.operation",
+        event = "s3_operation",
+        component = "storage",
+        operation = operation_name,
+        "rustfs.operation" = operation_name
+    );
+    let in_flight = gauge!(METRIC_REQUESTS_IN_FLIGHT, "operation" => operation_name);
+    in_flight.increment(1.0);
+    let _in_flight = InFlightRequest { gauge: in_flight };
+    let started_at = Instant::now();
+    let result = future.instrument(span.clone()).await;
+
+    let result_class = if result.is_ok() {
+        ResultClass::Success
+    } else {
+        ResultClass::Error
+    };
+    span.set_status(match result_class {
+        ResultClass::Success => Status::Ok,
+        ResultClass::Error => Status::error(ErrorClass::Internal.as_str()),
+    });
+    counter!(METRIC_REQUESTS_TOTAL, "operation" => operation_name, "result" => result_class.as_str()).increment(1);
+    histogram!(METRIC_REQUEST_DURATION_SECONDS, "operation" => operation_name, "result" => result_class.as_str())
+        .record(started_at.elapsed().as_secs_f64());
+    if result_class == ResultClass::Error {
+        counter!(METRIC_REQUEST_FAILURES_TOTAL, "operation" => operation_name, "error.type" => ErrorClass::Internal.as_str())
+            .increment(1);
+    }
+    result
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use metrics::with_local_recorder;
+    use metrics_util::debugging::{DebugValue, DebuggingRecorder};
+    use std::collections::HashSet;
+    use std::task::Poll;
+
+    fn capture_operation_metrics(result: Result<(), ()>) -> Vec<(String, HashSet<String>)> {
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        let runtime = tokio::runtime::Runtime::new().expect("create test runtime");
+        with_local_recorder(&recorder, || {
+            let observed = runtime.block_on(observe_operation(Operation::PutObject, async { result }));
+            assert_eq!(observed, result);
+        });
+        snapshotter
+            .snapshot()
+            .into_vec()
+            .into_iter()
+            .map(|(composite, _, _, _)| {
+                (
+                    composite.key().name().to_string(),
+                    composite.key().labels().map(|label| label.key().to_string()).collect(),
+                )
+            })
+            .collect()
+    }
+
+    #[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");
+    }
+
+    #[test]
+    fn operation_metrics_use_only_bounded_safe_labels() {
+        let metrics = capture_operation_metrics(Err(()));
+        let labels_for = |name| {
+            metrics
+                .iter()
+                .filter(|(metric_name, _)| metric_name == name)
+                .map(|(_, labels)| labels)
+                .collect::<Vec<_>>()
+        };
+
+        let operation_result = HashSet::from(["operation".to_string(), "result".to_string()]);
+        let operation_error = HashSet::from(["operation".to_string(), "error.type".to_string()]);
+        let operation_only = HashSet::from(["operation".to_string()]);
+        assert_eq!(labels_for(METRIC_REQUESTS_TOTAL), vec![&operation_result]);
+        assert_eq!(labels_for(METRIC_REQUEST_FAILURES_TOTAL), vec![&operation_error]);
+        assert_eq!(labels_for(METRIC_REQUEST_DURATION_SECONDS), vec![&operation_result]);
+        assert_eq!(labels_for(METRIC_REQUESTS_IN_FLIGHT), vec![&operation_only]);
+    }
+
+    #[test]
+    fn canceled_operation_releases_in_flight_gauge() {
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        let runtime = tokio::runtime::Runtime::new().expect("create test runtime");
+        with_local_recorder(&recorder, || {
+            runtime.block_on(async {
+                let operation = std::pin::pin!(observe_operation(Operation::GetObject, std::future::pending::<Result<(), ()>>()));
+                assert!(matches!(futures_util::poll!(operation), Poll::Pending));
+            });
+        });
+
+        let gauge = snapshotter
+            .snapshot()
+            .into_vec()
+            .into_iter()
+            .find_map(|(composite, _, _, value)| (composite.key().name() == METRIC_REQUESTS_IN_FLIGHT).then_some(value));
+        assert_eq!(gauge, Some(DebugValue::Gauge(0.0.into())));
+    }
+}

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<F>(future: F) -> tokio::task::JoinHandle<F::Output>
+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);
+    }
+}

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

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())));
+        .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()));
+    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(file_layer_opt.map(|layer| layer.with_filter(build_env_filter(logger_level, None))))
-        .with(stdout_layer_opt)
+        .with(stdout_layer_opt.map(|layer| layer.with_filter(build_env_filter(logger_level, None))))
         .with(tracer_layer)
         .with(otel_bridge)
         .with(metrics_layer)

crates/obs/src/telemetry/resource.rs

@@ -24,7 +24,9 @@ use opentelemetry::KeyValue;
 use opentelemetry_sdk::Resource;
 use opentelemetry_semantic_conventions::{
     SCHEMA_URL,
-    attribute::{DEPLOYMENT_ENVIRONMENT_NAME, NETWORK_LOCAL_ADDRESS, SERVICE_VERSION as OTEL_SERVICE_VERSION},
+    attribute::{
+        DEPLOYMENT_ENVIRONMENT_NAME, NETWORK_LOCAL_ADDRESS, SERVICE_INSTANCE_ID, SERVICE_VERSION as OTEL_SERVICE_VERSION,
+    },
 };
 use rustfs_config::{APP_NAME, ENVIRONMENT, SERVICE_VERSION};
 use rustfs_utils::get_local_ip_with_default;
@@ -44,21 +46,57 @@ use std::borrow::Cow;
 /// All attributes are attached to the resource using the semantic conventions
 /// schema URL to ensure compatibility with standard OTLP backends.
 pub(super) fn build_resource(config: &OtelConfig) -> Resource {
+    let mut attributes = vec![
+        KeyValue::new(
+            OTEL_SERVICE_VERSION,
+            Cow::Borrowed(config.service_version.as_deref().unwrap_or(SERVICE_VERSION)).to_string(),
+        ),
+        KeyValue::new(
+            DEPLOYMENT_ENVIRONMENT_NAME,
+            Cow::Borrowed(config.environment.as_deref().unwrap_or(ENVIRONMENT)).to_string(),
+        ),
+        KeyValue::new(NETWORK_LOCAL_ADDRESS, get_local_ip_with_default()),
+    ];
+    if let Some(instance_id) = config.instance_id.as_deref().filter(|value| !value.is_empty()) {
+        attributes.push(KeyValue::new(SERVICE_INSTANCE_ID, instance_id.to_string()));
+    }
+    if let Some(cluster_id) = config.cluster_id.as_deref().filter(|value| !value.is_empty()) {
+        attributes.push(KeyValue::new("rustfs.cluster.id", cluster_id.to_string()));
+    }
+
     Resource::builder()
         .with_service_name(Cow::Borrowed(config.service_name.as_deref().unwrap_or(APP_NAME)).to_string())
-        .with_schema_url(
+        .with_schema_url(attributes, SCHEMA_URL)
-            [
-                KeyValue::new(
-                    OTEL_SERVICE_VERSION,
-                    Cow::Borrowed(config.service_version.as_deref().unwrap_or(SERVICE_VERSION)).to_string(),
-                ),
-                KeyValue::new(
-                    DEPLOYMENT_ENVIRONMENT_NAME,
-                    Cow::Borrowed(config.environment.as_deref().unwrap_or(ENVIRONMENT)).to_string(),
-                ),
-                KeyValue::new(NETWORK_LOCAL_ADDRESS, get_local_ip_with_default()),
-            ],
-            SCHEMA_URL,
-        )
         .build()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use opentelemetry::Key;
+
+    #[test]
+    fn resource_uses_explicit_identity_without_fabricating_one() {
+        let resource = build_resource(&OtelConfig::default());
+        assert!(resource.get(&Key::new(SERVICE_INSTANCE_ID)).is_none());
+        assert!(resource.get(&Key::new("rustfs.cluster.id")).is_none());
+
+        let resource = build_resource(&OtelConfig {
+            instance_id: Some("node-a".to_string()),
+            cluster_id: Some("cluster-a".to_string()),
+            ..OtelConfig::default()
+        });
+        assert_eq!(
+            resource
+                .get(&Key::new(SERVICE_INSTANCE_ID))
+                .map(|value| value.as_str().to_string()),
+            Some("node-a".to_string())
+        );
+        assert_eq!(
+            resource
+                .get(&Key::new("rustfs.cluster.id"))
+                .map(|value| value.as_str().to_string()),
+            Some("cluster-a".to_string())
+        );
+    }
+}

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 }

crates/scanner/src/scanner_io.rs

@@ -786,7 +786,7 @@ impl ScannerIO for ECStore {
                 let (tx, mut rx) = mpsc::channel::<DataUsageCache>(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;

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;

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."));
+        }
+    }
+}

rustfs/src/app/multipart_usecase.rs

@@ -92,7 +92,7 @@ use std::str::FromStr;
 use std::sync::Arc;
 use tokio::sync::RwLock;
 use tokio_util::io::StreamReader;
-use tracing::{instrument, warn};
+use tracing::{Instrument as _, instrument, warn};
 use urlencoding::encode;
 use uuid::Uuid;
 
@@ -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, &current_opts).await {
+        let previous_current_size = match store
+            .get_object_info(&bucket, &key, &current_opts)
+            .instrument(rustfs_obs::stage_span(rustfs_obs::Stage::Metadata))
+            .await
+        {
             Ok(existing_obj_info) => {
                 validate_existing_object_lock_for_write(&existing_obj_info, &current_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();
@@ -484,6 +494,7 @@ impl DefaultMultipartUsecase {
         let obj_info = store
             .clone()
             .complete_multipart_upload(&bucket, &key, &upload_id, uploaded_parts, &opts)
+            .instrument(rustfs_obs::stage_span(rustfs_obs::Stage::EcWrite))
             .await
             .map_err(ApiError::from)?;
         let _ = invalidate_object_data_cache_after_complete_multipart_success(&cache_adapter, &bucket, &key).await;
@@ -755,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)?;
 
@@ -826,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)?;
 
@@ -967,6 +980,8 @@ 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)?;
 
@@ -1132,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)?;
 
@@ -1152,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)?;
 
@@ -1208,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;
@@ -1333,6 +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::EcWrite))
+            .instrument(rustfs_obs::stream_span(rustfs_obs::StreamDirection::Copy, actual_size, 0))
             .await
             .map_err(ApiError::from)?;
 

rustfs/src/app/object_usecase.rs

@@ -187,7 +187,7 @@ use tokio::io::{AsyncRead, ReadBuf};
 use tokio::sync::{OwnedSemaphorePermit, RwLock};
 use tokio_tar::Archive;
 use tokio_util::io::{ReaderStream, StreamReader};
-use tracing::{debug, error, instrument, warn};
+use tracing::{Instrument as _, debug, error, instrument, warn};
 use uuid::Uuid;
 
 use super::storage_api::object_usecase::{
@@ -2784,6 +2784,7 @@ impl DefaultObjectUsecase {
     ) -> S3Result<GetObjectReadSetup> {
         let reader = store
             .get_object_reader(bucket, key, rs.clone(), h, opts)
+            .instrument(rustfs_obs::stage_span(rustfs_obs::Stage::EcRead))
             .await
             .map_err(map_get_object_reader_error)?;
         if let Some(read_stage_start) = read_stage_start {
@@ -3675,7 +3676,10 @@ impl DefaultObjectUsecase {
             );
             let previous_current_info = {
                 crate::hp_guard!("S3::put_object_prelookup");
-                store.get_object_info(&bucket, &key, &current_opts).await
+                store
+                    .get_object_info(&bucket, &key, &current_opts)
+                    .instrument(rustfs_obs::stage_span(rustfs_obs::Stage::Metadata))
+                    .await
             };
             Some(match previous_current_info {
                 Ok(existing_obj_info) => {
@@ -3876,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)
         {

rustfs/src/storage/ecfs.rs

@@ -40,6 +40,7 @@ use http::StatusCode;
 use metrics::{counter, histogram};
 use rustfs_io_metrics::record_s3_op;
 use rustfs_madmin::{SITE_REPL_API_VERSION, SRBucketMeta};
+use rustfs_obs::{Operation as ObsOperation, observe_operation};
 use rustfs_s3_ops::S3Operation;
 use rustfs_targets::EventName;
 use rustfs_utils::http::headers::{
@@ -267,7 +268,7 @@ impl S3 for FS {
         req: S3Request<AbortMultipartUploadInput>,
     ) -> S3Result<S3Response<AbortMultipartUploadOutput>> {
         let usecase = s3_api::multipart_usecase_for(self);
-        usecase.execute_abort_multipart_upload(req).await
+        observe_operation(ObsOperation::AbortMultipartUpload, usecase.execute_abort_multipart_upload(req)).await
     }
 
     #[instrument(level = "debug", skip(self, req))]
@@ -277,7 +278,11 @@ impl S3 for FS {
     ) -> S3Result<S3Response<CompleteMultipartUploadOutput>> {
         crate::hp_guard!("S3::complete_multipart_upload");
         let usecase = s3_api::multipart_usecase_for(self);
-        Box::pin(usecase.execute_complete_multipart_upload(req)).await
+        observe_operation(
+            ObsOperation::CompleteMultipartUpload,
+            Box::pin(usecase.execute_complete_multipart_upload(req)),
+        )
+        .await
     }
 
     /// Copy an object from one location to another
@@ -294,7 +299,7 @@ impl S3 for FS {
     )]
     async fn create_bucket(&self, req: S3Request<CreateBucketInput>) -> S3Result<S3Response<CreateBucketOutput>> {
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_create_bucket(req).await
+        observe_operation(ObsOperation::CreateBucket, usecase.execute_create_bucket(req)).await
     }
 
     #[instrument(level = "debug", skip(self, req))]
@@ -304,14 +309,14 @@ impl S3 for FS {
     ) -> S3Result<S3Response<CreateMultipartUploadOutput>> {
         crate::hp_guard!("S3::create_multipart_upload");
         let usecase = s3_api::multipart_usecase_for(self);
-        usecase.execute_create_multipart_upload(req).await
+        observe_operation(ObsOperation::CreateMultipartUpload, usecase.execute_create_multipart_upload(req)).await
     }
 
     /// Delete a bucket
     #[instrument(level = "debug", skip(self, req))]
     async fn delete_bucket(&self, req: S3Request<DeleteBucketInput>) -> S3Result<S3Response<DeleteBucketOutput>> {
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_delete_bucket(req).await
+        observe_operation(ObsOperation::DeleteBucket, usecase.execute_delete_bucket(req)).await
     }
 
     #[instrument(level = "debug", skip(self))]
@@ -396,7 +401,7 @@ impl S3 for FS {
     async fn delete_object(&self, req: S3Request<DeleteObjectInput>) -> S3Result<S3Response<DeleteObjectOutput>> {
         crate::hp_guard!("S3::delete_object");
         let usecase = s3_api::object_usecase_for(self);
-        Box::pin(usecase.execute_delete_object(req)).await
+        observe_operation(ObsOperation::DeleteObject, Box::pin(usecase.execute_delete_object(req))).await
     }
 
     #[instrument(level = "debug", skip(self))]
@@ -682,7 +687,7 @@ impl S3 for FS {
     async fn get_object(&self, req: S3Request<GetObjectInput>) -> S3Result<S3Response<GetObjectOutput>> {
         crate::hp_guard!("S3::get_object");
         let usecase = s3_api::object_usecase_for(self);
-        Box::pin(usecase.execute_get_object(req)).await
+        observe_operation(ObsOperation::GetObject, Box::pin(usecase.execute_get_object(req))).await
     }
 
     async fn get_object_acl(&self, req: S3Request<GetObjectAclInput>) -> S3Result<S3Response<GetObjectAclOutput>> {
@@ -960,14 +965,14 @@ impl S3 for FS {
     #[instrument(level = "debug", skip(self, req))]
     async fn head_bucket(&self, req: S3Request<HeadBucketInput>) -> S3Result<S3Response<HeadBucketOutput>> {
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_head_bucket(req).await
+        observe_operation(ObsOperation::HeadBucket, usecase.execute_head_bucket(req)).await
     }
 
     #[instrument(level = "debug", skip(self, req))]
     async fn head_object(&self, req: S3Request<HeadObjectInput>) -> S3Result<S3Response<HeadObjectOutput>> {
         crate::hp_guard!("S3::head_object");
         let usecase = s3_api::object_usecase_for(self);
-        usecase.execute_head_object(req).await
+        observe_operation(ObsOperation::HeadObject, usecase.execute_head_object(req)).await
     }
 
     #[instrument(level = "debug", skip(self))]
@@ -975,7 +980,7 @@ impl S3 for FS {
         // List buckets not associated with a bucket, give it bucket label "*" to denote "all".
         record_s3_op(S3Operation::ListBuckets);
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_list_buckets(req).await
+        observe_operation(ObsOperation::ListBuckets, usecase.execute_list_buckets(req)).await
     }
 
     async fn list_multipart_uploads(
@@ -1000,7 +1005,7 @@ impl S3 for FS {
     async fn list_objects(&self, req: S3Request<ListObjectsInput>) -> S3Result<S3Response<ListObjectsOutput>> {
         record_s3_op(S3Operation::ListObjects);
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_list_objects(req).await
+        observe_operation(ObsOperation::ListObjects, usecase.execute_list_objects(req)).await
     }
 
     #[instrument(level = "debug", skip(self, req))]
@@ -1008,7 +1013,7 @@ impl S3 for FS {
         crate::hp_guard!("S3::list_objects_v2");
         record_s3_op(S3Operation::ListObjectsV2);
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_list_objects_v2(req).await
+        observe_operation(ObsOperation::ListObjectsV2, usecase.execute_list_objects_v2(req)).await
     }
 
     #[instrument(level = "debug", skip(self, req))]
@@ -1215,7 +1220,7 @@ impl S3 for FS {
     async fn put_object(&self, req: S3Request<PutObjectInput>) -> S3Result<S3Response<PutObjectOutput>> {
         crate::hp_guard!("S3::put_object");
         let usecase = s3_api::object_usecase_for(self);
-        Box::pin(usecase.execute_put_object(self, req)).await
+        observe_operation(ObsOperation::PutObject, Box::pin(usecase.execute_put_object(self, req))).await
     }
 
     async fn put_object_acl(&self, req: S3Request<PutObjectAclInput>) -> S3Result<S3Response<PutObjectAclOutput>> {
@@ -1577,13 +1582,13 @@ impl S3 for FS {
         crate::hp_guard!("S3::upload_part");
         record_s3_op(S3Operation::UploadPart);
         let usecase = s3_api::multipart_usecase_for(self);
-        usecase.execute_upload_part(req).await
+        observe_operation(ObsOperation::UploadPart, usecase.execute_upload_part(req)).await
     }
 
     #[instrument(level = "debug", skip(self, req))]
     async fn upload_part_copy(&self, req: S3Request<UploadPartCopyInput>) -> S3Result<S3Response<UploadPartCopyOutput>> {
         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
     }
 }

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,
+        CORSConfiguration, CORSRule, CopySource, DefaultRetention, DeleteObjectTaggingInput, Delimiter, FilterRule,
-        GetBucketAclInput, GetObjectAclInput, GetObjectLegalHoldInput, GetObjectRetentionInput, GetObjectTaggingInput,
+        FilterRuleName, GetBucketAclInput, GetObjectAclInput, GetObjectLegalHoldInput, GetObjectRetentionInput,
-        LambdaFunctionConfiguration, NotificationConfigurationFilter, ObjectLockConfiguration, ObjectLockEnabled,
+        GetObjectTaggingInput, LambdaFunctionConfiguration, NotificationConfigurationFilter, ObjectLockConfiguration,
-        ObjectLockLegalHold, ObjectLockLegalHoldStatus, ObjectLockRetention, ObjectLockRetentionMode, ObjectLockRule,
+        ObjectLockEnabled, ObjectLockLegalHold, ObjectLockLegalHoldStatus, ObjectLockRetention, ObjectLockRetentionMode,
-        PutBucketAclInput, PutObjectAclInput, PutObjectLegalHoldInput, PutObjectLockConfigurationInput, PutObjectRetentionInput,
+        ObjectLockRule, PutBucketAclInput, PutObjectAclInput, PutObjectLegalHoldInput, PutObjectLockConfigurationInput,
-        PutObjectTaggingInput, QueueConfiguration, S3KeyFilter, Tag, Tagging, TopicConfiguration,
+        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<T>(input: T, method: Method) -> S3Request<T> {
         S3Request {
@@ -59,6 +65,27 @@ mod tests {
         }
     }
 
+    #[derive(Clone)]
+    struct OperationSpanLayer {
+        seen: Arc<AtomicBool>,
+    }
+
+    impl<S> Layer<S> 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();

rustfs/src/storage/request_context.rs

@@ -188,7 +188,7 @@ pub fn spawn_traced<F>(fut: F)
 where
     F: std::future::Future<Output = ()> + Send + 'static,
 {
-    tokio::spawn(tracing::Instrument::instrument(fut, tracing::Span::current()));
+    rustfs_obs::spawn_traced(fut);
 }
 
 #[cfg(test)]