Add observability identities and operation spans

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,
-    ENV_OBS_ENDPOINT_LOGS_HEADERS, ENV_OBS_ENDPOINT_LOGS_TIMEOUT_MILLIS, ENV_OBS_ENDPOINT_METRICS_HEADERS,
-    ENV_OBS_ENDPOINT_METRICS_TIMEOUT_MILLIS, ENV_OBS_ENDPOINT_TIMEOUT_MILLIS, ENV_OBS_ENDPOINT_TRACES_HEADERS,
-    ENV_OBS_ENDPOINT_TRACES_TIMEOUT_MILLIS, ENV_OBS_ENVIRONMENT, ENV_OBS_LOG_CLEANUP_INTERVAL_SECONDS,
-    ENV_OBS_LOG_COMPRESS_OLD_FILES, ENV_OBS_LOG_COMPRESSED_FILE_RETENTION_DAYS, ENV_OBS_LOG_COMPRESSION_ALGORITHM,
-    ENV_OBS_LOG_DELETE_EMPTY_FILES, ENV_OBS_LOG_DIRECTORY, ENV_OBS_LOG_DRY_RUN, ENV_OBS_LOG_ENDPOINT,
-    ENV_OBS_LOG_EXCLUDE_PATTERNS, ENV_OBS_LOG_FILENAME, ENV_OBS_LOG_GZIP_COMPRESSION_LEVEL, 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,
+    DEFAULT_OBS_LOG_ZSTD_FALLBACK_TO_GZIP, DEFAULT_OBS_LOG_ZSTD_WORKERS, ENV_OBS_CLUSTER_ID, ENV_OBS_ENDPOINT,
+    ENV_OBS_ENDPOINT_HEADERS, ENV_OBS_ENDPOINT_LOGS_HEADERS, ENV_OBS_ENDPOINT_LOGS_TIMEOUT_MILLIS,
+    ENV_OBS_ENDPOINT_METRICS_HEADERS, ENV_OBS_ENDPOINT_METRICS_TIMEOUT_MILLIS, ENV_OBS_ENDPOINT_TIMEOUT_MILLIS,
+    ENV_OBS_ENDPOINT_TRACES_HEADERS, ENV_OBS_ENDPOINT_TRACES_TIMEOUT_MILLIS, ENV_OBS_ENVIRONMENT, ENV_OBS_INSTANCE_ID,
+    ENV_OBS_LOG_CLEANUP_INTERVAL_SECONDS, ENV_OBS_LOG_COMPRESS_OLD_FILES, ENV_OBS_LOG_COMPRESSED_FILE_RETENTION_DAYS,
+    ENV_OBS_LOG_COMPRESSION_ALGORITHM, ENV_OBS_LOG_DELETE_EMPTY_FILES, ENV_OBS_LOG_DIRECTORY, ENV_OBS_LOG_DRY_RUN,
+    ENV_OBS_LOG_ENDPOINT, ENV_OBS_LOG_EXCLUDE_PATTERNS, ENV_OBS_LOG_FILENAME, ENV_OBS_LOG_GZIP_COMPRESSION_LEVEL,
+    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,7 @@ mod error;
 mod global;
 mod logging;
 pub mod metrics;
+pub mod semconv;
 mod telemetry;
 
 pub use cleaner::*;
@@ -78,6 +79,7 @@ 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 telemetry::{OtelGuard, Recorder};
 
 // Dial9 Tokio runtime telemetry

crates/obs/src/semconv.rs

@@ -0,0 +1,265 @@
+//! 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 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,
+    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::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,
+}
+
+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())
+}
+
+/// 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
+    );
+    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::CompleteMultipartUpload.as_str(), "complete_multipart_upload");
+        assert_eq!(Stage::ReplicationRemote.as_str(), "replication_remote");
+        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/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(
-            [
-                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,
-        )
+        .with_schema_url(attributes, 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())
+        );
+    }
+}