Add Cosmos client-side observability layer

sdk/cosmos/.cspell.json

@@ -117,6 +117,7 @@
     "firewalled",
     "fixdate",
     "flamegraph",
+    "flatbuffer",
     "fmix",
     "fract",
     "francecentral",
@@ -175,6 +176,7 @@
     "MEMORYSTATUSEX",
     "Mgmt",
     "malloc",
+    "materializer",
     "mmap",
     "moka",
     "mpmc",
@@ -220,6 +222,7 @@
     "pgcosmos",
     "PIDS",
     "Pigw",
+    "pingpong",
     "pinning",
     "Pkrange",
     "pkranges",
@@ -256,12 +259,14 @@
     "rhost",
     "RNTBD",
     "rgby",
+    "rollup",
     "roundtripped",
     "roundtrips",
     "RTLD",
     "RUPM",
     "rwcache",
     "sbyte",
+    "semconv",
     "serviceunavailable",
     "serviceversion",
     "sess",
@@ -307,6 +312,7 @@
     "unavail",
     "uncollapsed",
     "uncontended",
+    "underspecified",
     "undrained",
     "unfaulted",
     "ungoverned",
@@ -345,6 +351,8 @@
     "westus",
     "widenings",
     "winget",
+    "workstream",
+    "workstreams",
     "writability",
     "xorshift",
     "xpart",

sdk/cosmos/azure_data_cosmos/CHANGELOG.md

@@ -4,6 +4,10 @@
 
 ### Features Added
 
+- Added a pluggable client-side diagnostics emission layer — the `DiagnosticsHandler` trait and ordered `DiagnosticsHandlerChain` (registered via `CosmosClientBuilder::with_diagnostics_handler`) — invoked once per operation (singleton and paginated, on success and failure) with the completed `DiagnosticsContext` plus an SDK-supplied `CosmosOperationContext`; the empty default chain is a zero-overhead no-op. ([#4789](https://github.com/Azure/azure-sdk-for-rust/pull/4789))
+- Added the `otel_metrics`-gated `CosmosMetricsHandler` (with `MetricsOptions`), emitting the stable `db.client.operation.duration` histogram plus opt-in development metrics (`operation.request_charge`, `response.returned_rows`); a no-op when no meter provider is registered. ([#4789](https://github.com/Azure/azure-sdk-for-rust/pull/4789))
+- Added tail-sampled emission handlers — `SamplingLogHandler` (rate-limited `tracing` line) and the `otel_tracing`-gated `CosmosTracingHandler` (backdated span tree) — that emit only for operations which fail or breach a configurable `DiagnosticsThresholds`. ([#4789](https://github.com/Azure/azure-sdk-for-rust/pull/4789))
+
 ### Breaking Changes
 
 ### Bugs Fixed

sdk/cosmos/azure_data_cosmos/Cargo.toml

@@ -20,6 +20,7 @@ azure_core = { workspace = true, default-features = false }
 azure_data_cosmos_driver = { version = "0.7.0", path = "../azure_data_cosmos_driver", default-features = false }
 base64.workspace = true
 futures.workspace = true
+opentelemetry = { workspace = true, optional = true, features = ["metrics"] }
 pin-project.workspace = true
 reqwest = { workspace = true, optional = true }
 serde.workspace = true
@@ -51,6 +52,10 @@ azure_data_cosmos_driver = { path = "../azure_data_cosmos_driver", default-featu
 ] }
 azure_identity.workspace = true
 clap.workspace = true
+# Enabled here (dev-only) so the metrics and tracing handler tests can drive
+# in-memory OpenTelemetry exporters. `testing` provides `InMemoryMetricExporter`
+# and `InMemorySpanExporter`.
+opentelemetry_sdk = { workspace = true, features = ["testing"] }
 reqwest = { workspace = true, features = ["http2"] }
 serde = { workspace = true, features = ["derive"] }
 serde_bytes.workspace = true
@@ -79,11 +84,17 @@ rustls = ["reqwest", "azure_data_cosmos_driver/rustls", "__tls"]
 native_tls = ["reqwest", "azure_data_cosmos_driver/native_tls", "__tls"]
 key_auth = [
 ] # Enables support for key-based authentication (Primary Keys and Resource Tokens)
+otel_tracing = [
+  "dep:opentelemetry",
+] # Enables the OpenTelemetry tracing diagnostics handler (off by default)
 hmac_rust = ["azure_core/hmac_rust"]
 hmac_openssl = ["azure_core/hmac_openssl"]
 fault_injection = [
   "azure_data_cosmos_driver/fault_injection",
 ] # Enables support for fault injection testing
+otel_metrics = [
+  "dep:opentelemetry",
+] # Enables the OpenTelemetry metrics handler (`CosmosMetricsHandler`). Off by default: with the feature disabled the metrics module is not compiled, so there is zero cost.
 preview_dtx = [
   "azure_data_cosmos_driver/preview_dtx",
 ] # Enables preview Distributed Transaction APIs. Disabled by default and not production-ready.
@@ -100,6 +111,8 @@ features = [
   "hmac_rust",
   "key_auth",
   "native_tls",
+  "otel_metrics",
+  "otel_tracing",
   "rustls",
 ]
 

sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs

@@ -3,6 +3,7 @@
 
 use crate::{
     clients::{offers_client, ClientContext},
+    diagnostics::CosmosOperationContext,
     feed::{ChangeFeedPageIterator, FeedRange, FeedScope, QueryItemIterator},
     models::TransactionalBatch,
     models::{BatchResponse, ChangeFeedItem, ItemResponse, ResourceResponse},
@@ -62,6 +63,16 @@ impl ContainerClient {
         })
     }
 
+    /// Builds the SDK-side [`CosmosOperationContext`] for this container's
+    /// operations, carrying the operation name plus the database and container
+    /// identity the driver context does not know.
+    fn operation_context(&self, operation_name: &'static str) -> CosmosOperationContext {
+        CosmosOperationContext::new()
+            .with_operation_name(operation_name)
+            .with_database_name(self.container_ref.database_name().to_string())
+            .with_container_name(self.container_ref.name().to_string())
+    }
+
     /// Reads the properties of the container.
     ///
     /// # Arguments
@@ -85,14 +96,15 @@ impl ContainerClient {
         let options = options.unwrap_or_default();
         let operation = CosmosOperation::read_container(self.container_ref.clone());
 
-        let driver_response = self
+        let driver_result = self
             .context
             .driver
             .execute_singleton_operation(operation, options.operation)
-            .await?;
+            .await;
 
         Ok(ResourceResponse::new(
-            crate::driver_bridge::driver_response_to_cosmos_response(driver_response),
+            self.context
+                .complete_result(driver_result, || self.operation_context("read_container"))?,
         ))
     }
 
@@ -138,14 +150,16 @@ impl ContainerClient {
         operation_options.content_response_on_write =
             Some(azure_data_cosmos_driver::options::ContentResponseOnWrite::Enabled);
 
-        let driver_response = self
+        let driver_result = self
             .context
             .driver
             .execute_singleton_operation(operation, operation_options)
-            .await?;
+            .await;
 
         Ok(ResourceResponse::new(
-            crate::driver_bridge::driver_response_to_cosmos_response(driver_response),
+            self.context.complete_result(driver_result, || {
+                self.operation_context("replace_container")
+            })?,
         ))
     }
 
@@ -161,10 +175,11 @@ impl ContainerClient {
     ) -> crate::Result<Option<ThroughputProperties>> {
         let options = options.unwrap_or_default();
         offers_client::find_offer(
-            &self.context.driver,
+            &self.context,
             self.container_ref.account(),
             self.container_ref.rid(),
             options.operation,
+            self.operation_context("read_throughput"),
         )
         .await
     }
@@ -202,11 +217,12 @@ impl ContainerClient {
         let options = options.unwrap_or_default();
 
         offers_client::begin_replace(
-            self.context.driver.clone(),
+            self.context.clone(),
             self.container_ref.account().clone(),
             self.container_ref.rid(),
             throughput,
             options.operation,
+            self.operation_context("replace_throughput"),
         )
         .await
     }
@@ -224,14 +240,15 @@ impl ContainerClient {
         let options = options.unwrap_or_default();
         let operation = CosmosOperation::delete_container(self.container_ref.clone());
 
-        let driver_response = self
+        let driver_result = self
             .context
             .driver
             .execute_singleton_operation(operation, options.operation)
-            .await?;
+            .await;
 
         Ok(ResourceResponse::new(
-            crate::driver_bridge::driver_response_to_cosmos_response(driver_response),
+            self.context
+                .complete_result(driver_result, || self.operation_context("delete_container"))?,
         ))
     }
 
@@ -322,15 +339,16 @@ impl ContainerClient {
         let operation = apply_item_options(operation, options.session_token, options.precondition);
 
         // Execute through the driver.
-        let driver_response = self
+        let driver_result = self
             .context
             .driver
             .execute_singleton_operation(operation, options.operation)
-            .await?;
+            .await;
 
         // Bridge the driver response to the SDK response type.
         Ok(ItemResponse::new(
-            crate::driver_bridge::driver_response_to_cosmos_response(driver_response),
+            self.context
+                .complete_result(driver_result, || self.operation_context("create_item"))?,
         ))
     }
 
@@ -420,15 +438,16 @@ impl ContainerClient {
         let operation = apply_item_options(operation, options.session_token, options.precondition);
 
         // Execute through the driver.
-        let driver_response = self
+        let driver_result = self
             .context
             .driver
             .execute_singleton_operation(operation, options.operation)
-            .await?;
+            .await;
 
         // Bridge the driver response to the SDK response type.
         Ok(ItemResponse::new(
-            crate::driver_bridge::driver_response_to_cosmos_response(driver_response),
+            self.context
+                .complete_result(driver_result, || self.operation_context("replace_item"))?,
         ))
     }
 
@@ -527,14 +546,15 @@ impl ContainerClient {
         // session token.
         let operation = apply_item_options(operation, options.session_token, None);
 
-        let driver_response = self
+        let driver_result = self
             .context
             .driver
             .execute_singleton_operation(operation, options.operation)
-            .await?;
+            .await;
 
         Ok(ItemResponse::new(
-            crate::driver_bridge::driver_response_to_cosmos_response(driver_response),
+            self.context
+                .complete_result(driver_result, || self.operation_context("patch_item"))?,
         ))
     }
 
@@ -628,15 +648,16 @@ impl ContainerClient {
         let operation = apply_item_options(operation, options.session_token, options.precondition);
 
         // Execute through the driver.
-        let driver_response = self
+        let driver_result = self
             .context
             .driver
             .execute_singleton_operation(operation, options.operation)
-            .await?;
+            .await;
 
         // Bridge the driver response to the SDK response type.
         Ok(ItemResponse::new(
-            crate::driver_bridge::driver_response_to_cosmos_response(driver_response),
+            self.context
+                .complete_result(driver_result, || self.operation_context("upsert_item"))?,
         ))
     }
 
@@ -688,15 +709,16 @@ impl ContainerClient {
         let operation = apply_item_options(operation, options.session_token, options.precondition);
 
         // Execute through the driver.
-        let driver_response = self
+        let driver_result = self
             .context
             .driver
             .execute_singleton_operation(operation, options.operation)
-            .await?;
+            .await;
 
         // Bridge the driver response to the SDK response type.
         Ok(ItemResponse::new(
-            crate::driver_bridge::driver_response_to_cosmos_response(driver_response),
+            self.context
+                .complete_result(driver_result, || self.operation_context("read_item"))?,
         ))
     }
 
@@ -740,15 +762,16 @@ impl ContainerClient {
         let operation = apply_item_options(operation, options.session_token, options.precondition);
 
         // Execute through the driver.
-        let driver_response = self
+        let driver_result = self
             .context
             .driver
             .execute_singleton_operation(operation, options.operation)
-            .await?;
+            .await;
 
         // Bridge the driver response to the SDK response type.
         Ok(ItemResponse::new(
-            crate::driver_bridge::driver_response_to_cosmos_response(driver_response),
+            self.context
+                .complete_result(driver_result, || self.operation_context("delete_item"))?,
         ))
     }
 
@@ -858,6 +881,8 @@ impl ContainerClient {
             Some(self.container_ref.clone()),
             plan,
             options.operation,
+            self.context.diagnostics_handlers.clone(),
+            self.operation_context("query_items"),
         ))
     }
 
@@ -951,6 +976,8 @@ impl ContainerClient {
             Some(self.container_ref.clone()),
             plan,
             options.operation,
+            self.context.diagnostics_handlers.clone(),
+            self.operation_context("query_change_feed"),
         ))
     }
 
@@ -1008,14 +1035,15 @@ impl ContainerClient {
             CosmosOperation::batch(self.container_ref.clone(), driver_pk).with_body(body);
         let operation = apply_batch_options(operation, &options);
 
-        let driver_response = self
+        let driver_result = self
             .context
             .driver
             .execute_singleton_operation(operation, options.operation)
-            .await?;
+            .await;
 
         Ok(BatchResponse::new(
-            crate::driver_bridge::driver_response_to_cosmos_response(driver_response),
+            self.context
+                .complete_result(driver_result, || self.operation_context("execute_batch"))?,
         ))
     }