diff --git a/Cargo.lock b/Cargo.lock index 1505f2528..14384120f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9592,6 +9592,7 @@ dependencies = [ "jiff", "libc", "metrics", + "metrics-util", "num_cpus", "nvml-wrapper", "opentelemetry", diff --git a/crates/config/src/observability/mod.rs b/crates/config/src/observability/mod.rs index 9dd2ba654..1f41b9741 100644 --- a/crates/config/src/observability/mod.rs +++ b/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"); diff --git a/crates/obs/Cargo.toml b/crates/obs/Cargo.toml index 454305bed..2a45fab57 100644 --- a/crates/obs/Cargo.toml +++ b/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 } diff --git a/crates/obs/README.md b/crates/obs/README.md index 38cf734bd..c763c2113 100644 --- a/crates/obs/README.md +++ b/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 `-.rustfs.log` (e.g., `20231027103001.123456-0.rustfs.log`); with `prefix`, they look like `rustfs.log.-` (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): diff --git a/crates/obs/src/config.rs b/crates/obs/src/config.rs index ac067a50a..46b862408 100644 --- a/crates/obs/src/config.rs +++ b/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, /// Deployment environment tag, e.g. `production` or `development`. pub environment: Option, + /// Stable identity for this process instance. Empty values are omitted. + pub instance_id: Option, + /// Stable RustFS deployment identity. Empty values are omitted. + pub cluster_id: Option, // ── 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), diff --git a/crates/obs/src/lib.rs b/crates/obs/src/lib.rs index b2919d9b2..1e2640c9a 100644 --- a/crates/obs/src/lib.rs +++ b/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 diff --git a/crates/obs/src/semconv.rs b/crates/obs/src/semconv.rs new file mode 100644 index 000000000..b44e08043 --- /dev/null +++ b/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(operation: Operation, future: F) -> Result +where + F: Future>, +{ + 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)> { + 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::>() + }; + + 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::>())); + 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()))); + } +} diff --git a/crates/obs/src/telemetry/resource.rs b/crates/obs/src/telemetry/resource.rs index a323047f9..a4758eac7 100644 --- a/crates/obs/src/telemetry/resource.rs +++ b/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()) + ); + } +} diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 593b8d56a..ea737b5b3 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/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; @@ -484,6 +484,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; @@ -967,6 +968,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)) .await .map_err(ApiError::from)?; @@ -1333,6 +1335,7 @@ 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)) .await .map_err(ApiError::from)?; diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index d58db269a..45306f2a6 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/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 { 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 { diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index 7fdb63a5a..e84df936e 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/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, ) -> S3Result> { 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> { 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) -> S3Result> { 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> { 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) -> S3Result> { 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) -> S3Result> { 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) -> S3Result> { 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) -> S3Result> { @@ -960,14 +965,14 @@ impl S3 for FS { #[instrument(level = "debug", skip(self, req))] async fn head_bucket(&self, req: S3Request) -> S3Result> { 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) -> S3Result> { 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) -> S3Result> { 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) -> S3Result> { 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) -> S3Result> { @@ -1577,7 +1582,7 @@ 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))]