From 67221e7b0028b35d7eddf011e3c82c74b48ff537 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Knut=20Olav=20L=C3=B8ite?= Date: Sat, 8 Aug 2026 08:28:10 +0200 Subject: [PATCH 1/2] chore(spanner): implement built-in attempt metrics and server-timing recording - Implement SpannerMetricsInterceptor to record attempt latencies, attempt count, and server-timing metrics (GFE and AFE latencies) on RPC attempt completion. - Inject x-goog-spanner-enable-afe-server-timing request header on outgoing attempts, respecting SPANNER_DISABLE_AFE_SERVER_TIMING. - Normalize the method metric attribute to "Spanner." with zero-allocation static matching across operation and attempt metrics. - Fix GFE and AFE connectivity error recording to prevent false positive AFE errors on non-DirectPath traffic. - Disable Cloud Monitoring exporter initialization for plaintext mock/test endpoints, Omni, and emulator environments. --- src/gax-internal/src/attempt_interceptor.rs | 238 +++++- src/gax-internal/src/grpc.rs | 78 +- src/gax-internal/tests/grpc_retry_loop.rs | 58 ++ src/spanner/src/client.rs | 42 +- src/spanner/src/observability/exporter.rs | 34 +- src/spanner/src/observability/metrics.rs | 892 +++++++++++++++++--- src/spanner/src/request_id.rs | 4 +- src/spanner/src/session_maintainer.rs | 4 +- 8 files changed, 1187 insertions(+), 163 deletions(-) diff --git a/src/gax-internal/src/attempt_interceptor.rs b/src/gax-internal/src/attempt_interceptor.rs index 8b30fe8aa3..a2e5c80750 100644 --- a/src/gax-internal/src/attempt_interceptor.rs +++ b/src/gax-internal/src/attempt_interceptor.rs @@ -14,37 +14,251 @@ //! Types and traits for intercepting and modifying outgoing RPC attempts. +use google_cloud_gax::error::Error; +use google_cloud_gax::options::RequestOptions; use http::HeaderMap; +use std::fmt::Debug; use std::sync::Arc; +use std::time::Instant; -/// A callback invoked on every RPC attempt, allowing modification of gRPC headers. -/// The callback receives the header map and the current 1-based attempt number. -pub trait AttemptInterceptor: std::fmt::Debug + Send + Sync { +/// A callback invoked on outgoing RPC attempts, allowing modification of gRPC headers and tracking attempt lifecycle. +/// +/// # Unary vs. Streaming RPC Lifecycles +/// - [`intercept`](Self::intercept): Invoked for **all** outgoing RPCs (unary and streaming) before transmission, +/// allowing headers (such as authentication or request IDs) to be modified. +/// - [`on_attempt_start`](Self::on_attempt_start) and [`on_attempt_complete`](Self::on_attempt_complete): Lifecycle +/// hooks invoked specifically for **unary RPC attempts** managed by the GAX retry loop, allowing attempt +/// duration, response headers, and attempt outcomes to be measured. +/// +/// Streaming RPCs have lifecycles that extend across stream consumption beyond the initial request dispatch; +/// their requests invoke [`intercept`](Self::intercept) during stream establishment, while stream iteration +/// and retries are managed by the higher-level streaming layer. +pub trait AttemptInterceptor: Debug + Send + Sync { /// Intercepts and modifies the headers of an outgoing RPC attempt. /// - /// `headers` is the mutable map of headers to be sent with the request. - /// `attempt` is the 1-based attempt number for the current RPC. - fn intercept(&self, headers: &mut HeaderMap, attempt: u32); + /// This method is invoked for all outgoing RPCs (both unary and streaming). + /// + /// * `headers`: The mutable map of headers to be sent with the request. + /// * `attempt`: The 1-based attempt number for the current RPC. + fn intercept(&self, _headers: &mut HeaderMap, _attempt: u32) {} + + /// Callback invoked before a unary RPC attempt is dispatched by the GAX retry loop. + /// + /// Allows header mutation and returns the start [`Instant`] of the attempt. + /// The default implementation delegates to [`self.intercept(headers, attempt)`](Self::intercept) + /// and returns [`Instant::now()`]. + /// + /// * `method`: The gRPC method path (e.g. `"/google.spanner.v1.Spanner/ExecuteSql"`). + /// * `attempt`: The 1-based attempt number for the current RPC. + /// * `headers`: The mutable map of headers to be sent with the request. + /// * `options`: The request options associated with the RPC. + fn on_attempt_start( + &self, + _method: &str, + attempt: u32, + headers: &mut HeaderMap, + _options: &RequestOptions, + ) -> Instant { + self.intercept(headers, attempt); + Instant::now() + } + + /// Callback invoked when a unary RPC attempt completes (either successfully or with an error). + /// + /// * `method`: The gRPC method path (e.g. `"/google.spanner.v1.Spanner/ExecuteSql"`). + /// * `attempt`: The 1-based attempt number for the RPC. + /// * `start_time`: The instant returned by [`on_attempt_start`](Self::on_attempt_start). + /// * `response_headers`: The response metadata headers returned by the server, if available. + /// * `error`: The error returned by this attempt, if the attempt failed. + /// * `options`: The request options associated with the RPC. + fn on_attempt_complete( + &self, + _method: &str, + _attempt: u32, + _start_time: Instant, + _response_headers: Option<&HeaderMap>, + _error: Option<&Error>, + _options: &RequestOptions, + ) { + } } impl AttemptInterceptor for Vec> { - fn intercept(&self, headers: &mut HeaderMap, attempt: u32) { + /// Callback invoked before a unary RPC attempt is dispatched. + /// + /// # Note + /// Any custom `Instant` returned by individual interceptors in the `Vec` is discarded. + /// The composite implementation always returns a fresh `Instant::now()`. + fn on_attempt_start( + &self, + method: &str, + attempt: u32, + headers: &mut HeaderMap, + options: &RequestOptions, + ) -> Instant { for interceptor in self { - interceptor.intercept(headers, attempt); + interceptor.on_attempt_start(method, attempt, headers, options); } + Instant::now() } -} -impl AttemptInterceptor for Option { fn intercept(&self, headers: &mut HeaderMap, attempt: u32) { - if let Some(interceptor) = self { + for interceptor in self { interceptor.intercept(headers, attempt); } } + + fn on_attempt_complete( + &self, + method: &str, + attempt: u32, + start_time: Instant, + response_headers: Option<&HeaderMap>, + error: Option<&Error>, + options: &RequestOptions, + ) { + for interceptor in self { + interceptor.on_attempt_complete( + method, + attempt, + start_time, + response_headers, + error, + options, + ); + } + } } impl AttemptInterceptor for Arc { + fn on_attempt_start( + &self, + method: &str, + attempt: u32, + headers: &mut HeaderMap, + options: &RequestOptions, + ) -> Instant { + (**self).on_attempt_start(method, attempt, headers, options) + } + fn intercept(&self, headers: &mut HeaderMap, attempt: u32) { - self.as_ref().intercept(headers, attempt); + (**self).intercept(headers, attempt); + } + + fn on_attempt_complete( + &self, + method: &str, + attempt: u32, + start_time: Instant, + response_headers: Option<&HeaderMap>, + error: Option<&Error>, + options: &RequestOptions, + ) { + (**self).on_attempt_complete( + method, + attempt, + start_time, + response_headers, + error, + options, + ); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicU32, Ordering}; + + fn assert_send() {} + fn assert_sync() {} + fn assert_debug() {} + fn assert_attempt_interceptor() {} + + #[test] + fn traits() { + assert_send::>(); + assert_sync::>(); + assert_debug::>(); + assert_attempt_interceptor::>(); + + assert_send::>>(); + assert_sync::>>(); + assert_debug::>>(); + assert_attempt_interceptor::>>(); + } + + #[derive(Debug, Default)] + struct MockInterceptor { + intercept_count: AtomicU32, + complete_count: AtomicU32, + } + + impl AttemptInterceptor for MockInterceptor { + fn intercept(&self, _headers: &mut HeaderMap, _attempt: u32) { + self.intercept_count.fetch_add(1, Ordering::SeqCst); + } + + fn on_attempt_complete( + &self, + _method: &str, + _attempt: u32, + _start_time: Instant, + _response_headers: Option<&HeaderMap>, + _error: Option<&Error>, + _options: &RequestOptions, + ) { + self.complete_count.fetch_add(1, Ordering::SeqCst); + } + } + + #[test] + fn default_on_attempt_start_calls_intercept() { + let interceptor = MockInterceptor::default(); + let mut headers = HeaderMap::new(); + let options = RequestOptions::default(); + let _start = interceptor.on_attempt_start("test_method", 1, &mut headers, &options); + assert_eq!(interceptor.intercept_count.load(Ordering::SeqCst), 1); + } + + #[test] + fn vec_interceptor_forwards_to_all() { + let first = Arc::new(MockInterceptor::default()); + let second = Arc::new(MockInterceptor::default()); + let interceptors: Vec> = vec![first.clone(), second.clone()]; + let mut headers = HeaderMap::new(); + let options = RequestOptions::default(); + + let start = interceptors.on_attempt_start("test_method", 1, &mut headers, &options); + assert_eq!(first.intercept_count.load(Ordering::SeqCst), 1); + assert_eq!(second.intercept_count.load(Ordering::SeqCst), 1); + + interceptors.on_attempt_complete("test_method", 1, start, Some(&headers), None, &options); + assert_eq!(first.complete_count.load(Ordering::SeqCst), 1); + assert_eq!(second.complete_count.load(Ordering::SeqCst), 1); + } + + #[test] + fn arc_interceptor_forwards_all_methods() { + let mock = Arc::new(MockInterceptor::default()); + let arc_interceptor: Arc = mock.clone(); + let mut headers = HeaderMap::new(); + let options = RequestOptions::default(); + + let start = arc_interceptor.on_attempt_start("test_method", 1, &mut headers, &options); + assert_eq!(mock.intercept_count.load(Ordering::SeqCst), 1); + + arc_interceptor.intercept(&mut headers, 2); + assert_eq!(mock.intercept_count.load(Ordering::SeqCst), 2); + + arc_interceptor.on_attempt_complete( + "test_method", + 1, + start, + Some(&headers), + None, + &options, + ); + assert_eq!(mock.complete_count.load(Ordering::SeqCst), 1); } } diff --git a/src/gax-internal/src/grpc.rs b/src/gax-internal/src/grpc.rs index 30cef486dd..9244b0f3a2 100644 --- a/src/gax-internal/src/grpc.rs +++ b/src/gax-internal/src/grpc.rs @@ -28,6 +28,7 @@ use crate::attempt_interceptor::AttemptInterceptor; use crate::observability::attributes::{self, keys::*, otel_status_codes}; use crate::universe_domain::DEFAULT_UNIVERSE_DOMAIN; use ::tonic::client::Grpc; +use ::tonic::metadata::MetadataMap; use ::tonic::transport::Channel; use from_status::to_gax_error; use futures::TryFutureExt; @@ -46,6 +47,7 @@ use http::HeaderMap; use opentelemetry_semantic_conventions::{attribute as otel_attr, trace as otel_trace}; use std::sync::Arc; use std::time::Duration; +use std::time::Instant; use transport_policies::TransportPolicies; // A tonic::transport::Channel always has a Buffer layer. @@ -193,7 +195,9 @@ impl Client { use ::tonic::IntoStreamingRequest; let headers = make_headers(api_client_header, request_params, &options)?; let mut headers = add_auth_headers(headers, &self.credentials).await?; - self.attempt_interceptor.intercept(&mut headers, 1); + if let Some(ref interceptor) = self.attempt_interceptor { + interceptor.intercept(&mut headers, 1); + } let metadata = tonic::MetadataMap::from_headers(headers); let request = ::tonic::Request::from_parts(metadata, extensions, request); let codec = tonic_prost::ProstCodec::::default(); @@ -259,7 +263,9 @@ impl Client { use ::tonic::IntoRequest; let headers = make_headers(api_client_header, request_params, &options)?; let mut headers = add_auth_headers(headers, &self.credentials).await?; - self.attempt_interceptor.intercept(&mut headers, 1); + if let Some(ref interceptor) = self.attempt_interceptor { + interceptor.intercept(&mut headers, 1); + } let metadata = tonic::MetadataMap::from_headers(headers); let mut request = ::tonic::Request::from_parts(metadata, extensions, request); if let Some(timeout) = crate::options::resolve_effective_timeout( @@ -394,8 +400,8 @@ impl Client { let mut headers = add_auth_headers(headers, &self.credentials).await?; crate::observability::propagation::inject_context(&span, &mut headers); - self.attempt_interceptor - .intercept(&mut headers, prior_attempt_count as u32 + 1); + let attempt_number = prior_attempt_count as u32 + 1; + let start_time = self.on_attempt_start(path.path(), attempt_number, &mut headers, options); let metadata = tonic::MetadataMap::from_headers(headers); let mut request = ::tonic::Request::from_parts(metadata, extensions, request); @@ -415,7 +421,9 @@ impl Client { recorder.on_grpc_request(&path); } - let pending = inner.unary(request, path, codec).map_err(to_gax_error); + let pending = inner + .unary(request, path.clone(), codec) + .map_err(to_gax_error); use crate::observability::{WithTransportLogging, WithTransportMetric, WithTransportSpan}; @@ -424,10 +432,68 @@ impl Client { let pending = WithTransportLogging::new(pending); let pending = WithTransportSpan::new(span, pending); - if let Some(recorder) = crate::observability::RequestRecorder::current() { + let result = if let Some(recorder) = crate::observability::RequestRecorder::current() { recorder.scope(pending).await } else { pending.await + }; + + self.on_attempt_complete(path.path(), attempt_number, start_time, result, options) + } + + #[inline] + fn on_attempt_start( + &self, + method: &str, + attempt: u32, + headers: &mut HeaderMap, + options: &RequestOptions, + ) -> Option { + self.attempt_interceptor + .as_ref() + .map(|interceptor| interceptor.on_attempt_start(method, attempt, headers, options)) + } + + #[inline] + fn on_attempt_complete( + &self, + method: &str, + attempt: u32, + start_time: Option, + result: Result<::tonic::Response>, + options: &RequestOptions, + ) -> Result<::tonic::Response> { + if let (Some(interceptor), Some(start_time)) = (&self.attempt_interceptor, start_time) { + match result { + Ok(response) => { + let (metadata, message, extensions) = response.into_parts(); + let headers = metadata.into_headers(); + interceptor.on_attempt_complete( + method, + attempt, + start_time, + Some(&headers), + None, + options, + ); + let metadata = MetadataMap::from_headers(headers); + let response = ::tonic::Response::from_parts(metadata, message, extensions); + Ok(response) + } + Err(error) => { + interceptor.on_attempt_complete( + method, + attempt, + start_time, + error.http_headers(), + Some(&error), + options, + ); + Err(error) + } + } + } else { + result } } diff --git a/src/gax-internal/tests/grpc_retry_loop.rs b/src/gax-internal/tests/grpc_retry_loop.rs index f337523de4..ba859205b5 100644 --- a/src/gax-internal/tests/grpc_retry_loop.rs +++ b/src/gax-internal/tests/grpc_retry_loop.rs @@ -16,6 +16,7 @@ mod tests { use google_cloud_auth::credentials::{Credentials, anonymous::Builder as Anonymous}; use google_cloud_gax::backoff_policy::BackoffPolicy; + use google_cloud_gax::error::Error; use google_cloud_gax::exponential_backoff::ExponentialBackoffBuilder; use google_cloud_gax::options::RequestOptions; use google_cloud_gax::retry_policy::{Aip194Strict, RetryPolicyExt}; @@ -26,6 +27,7 @@ mod tests { use grpc_server::{builder, google, start_fixed_responses}; use http::HeaderMap; use std::sync::{Arc, Mutex}; + use std::time::Instant; fn test_credentials() -> Credentials { Anonymous::new().build() @@ -131,6 +133,62 @@ mod tests { Ok(()) } + #[tokio::test] + async fn interceptor_on_attempt_complete() -> anyhow::Result<()> { + #[derive(Debug, Default)] + struct AttemptCompletionTracker { + completed_attempts: Mutex>, + } + impl AttemptInterceptor for AttemptCompletionTracker { + fn intercept(&self, _headers: &mut HeaderMap, _attempt: u32) {} + + fn on_attempt_complete( + &self, + method: &str, + attempt: u32, + _start_time: Instant, + _response_headers: Option<&HeaderMap>, + error: Option<&Error>, + _options: &RequestOptions, + ) { + self.completed_attempts.lock().expect("lock failed").push(( + method.to_string(), + attempt, + error.is_none(), + )); + } + } + + let tracker = Arc::new(AttemptCompletionTracker::default()); + let (endpoint, _server) = + start_fixed_responses(vec![transient(), transient(), success()]).await?; + + let mut config = ClientConfig::default(); + config.cred = Some(test_credentials()); + config.endpoint = Some(endpoint); + config.backoff_policy = Some(Arc::new(test_backoff())); + + let mut client = grpc::Client::new(config, "https://test-only.googleapis.com").await?; + client.set_attempt_interceptor(tracker.clone()); + let _response = send_request(client, "interceptor_on_attempt_complete").await?; + + let completed = tracker + .completed_attempts + .lock() + .expect("lock failed") + .clone(); + assert_eq!( + completed, + vec![ + ("/google.test.v1.EchoService/Echo".to_string(), 1, false), + ("/google.test.v1.EchoService/Echo".to_string(), 2, false), + ("/google.test.v1.EchoService/Echo".to_string(), 3, true), + ] + ); + + Ok(()) + } + fn success() -> tonic::Result> { Ok(tonic::Response::new(EchoResponse { message: "success!".into(), diff --git a/src/spanner/src/client.rs b/src/spanner/src/client.rs index 6e90743797..7ca46301a6 100644 --- a/src/spanner/src/client.rs +++ b/src/spanner/src/client.rs @@ -18,10 +18,14 @@ use crate::model::{ ExecuteBatchDmlRequest, ExecuteBatchDmlResponse, ExecuteSqlRequest, PartitionQueryRequest, PartitionReadRequest, PartitionResponse, RollbackRequest, Session, Transaction, }; +use crate::observability::Observability; +#[cfg(feature = "_experimental-builtin-metrics")] +use crate::observability::metrics::SpannerMetricsInterceptor; use crate::omni::{InstanceType, is_plaintext_endpoint}; use crate::request_id::RequestIdCreator; use crate::request_id_interceptor::{REQUEST_ID_HEADER, SpannerRequestIdInterceptor}; use crate::server_streaming::builder; +use gaxi::attempt_interceptor::AttemptInterceptor; use gaxi::options::{ClientConfig, Credentials}; use google_cloud_auth::credentials::anonymous; use google_cloud_gax::client_builder::ClientBuilder as GaxClientBuilder; @@ -158,9 +162,11 @@ macro_rules! define_idempotent_rpc { request: $request_type, options: crate::RequestOptions, channel_hint: usize, - o11y: &crate::observability::Observability, + o11y: &Arc, ) -> crate::Result<$response_type> { let options = self.attach_request_id(options, channel_hint); + #[cfg(feature = "_experimental-builtin-metrics")] + let options = options.insert_extension(o11y.clone()); o11y.trace_operation( $canonical_name, self.get_channel(channel_hint) @@ -502,9 +508,19 @@ impl Channel { pub(crate) async fn create(config: &ClientConfig) -> crate::ClientBuilderResult { let mut transport = crate::generated::gapic_dataplane::transport::Spanner::new(config.clone()).await?; - transport - .inner - .set_attempt_interceptor(Arc::new(SpannerRequestIdInterceptor)); + let request_id_interceptor: Arc = + Arc::new(SpannerRequestIdInterceptor); + + #[cfg(feature = "_experimental-builtin-metrics")] + let interceptor: Arc = Arc::new(vec![ + request_id_interceptor, + Arc::new(SpannerMetricsInterceptor), + ]); + + #[cfg(not(feature = "_experimental-builtin-metrics"))] + let interceptor: Arc = request_id_interceptor; + + transport.inner.set_attempt_interceptor(interceptor); let grpc_client = transport.inner.clone(); let inner = if gaxi::options::tracing_enabled(config) { @@ -682,7 +698,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), - &crate::observability::Observability::disabled(), + &crate::observability::Observability::disabled_arc(), ) .await .expect("Failed to call create_session"); @@ -802,7 +818,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), - &crate::observability::Observability::disabled(), + &crate::observability::Observability::disabled_arc(), ) .await .expect("Failed to call create_session after transport error retry"); @@ -852,7 +868,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), - &crate::observability::Observability::disabled(), + &crate::observability::Observability::disabled_arc(), ) .await .expect("Failed to call execute_sql"); @@ -896,7 +912,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), - &crate::observability::Observability::disabled(), + &crate::observability::Observability::disabled_arc(), ) .await .expect("Failed to call execute_batch_dml"); @@ -935,7 +951,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), - &crate::observability::Observability::disabled(), + &crate::observability::Observability::disabled_arc(), ) .await .expect("Failed to call begin_transaction"); @@ -978,7 +994,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), - &crate::observability::Observability::disabled(), + &crate::observability::Observability::disabled_arc(), ) .await .expect("Failed to call commit"); @@ -1012,7 +1028,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), - &crate::observability::Observability::disabled(), + &crate::observability::Observability::disabled_arc(), ) .await .expect("Failed to call rollback"); @@ -1245,7 +1261,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), - &crate::observability::Observability::disabled(), + &crate::observability::Observability::disabled_arc(), ) .await .expect("Failed to call create_session"); @@ -1292,7 +1308,7 @@ mod tests { req, options, client.next_channel_hint(), - &crate::observability::Observability::disabled(), + &crate::observability::Observability::disabled_arc(), ) .await; diff --git a/src/spanner/src/observability/exporter.rs b/src/spanner/src/observability/exporter.rs index e5fbcb4ebc..0814e7e576 100644 --- a/src/spanner/src/observability/exporter.rs +++ b/src/spanner/src/observability/exporter.rs @@ -398,14 +398,14 @@ mod tests { use std::time::SystemTime; #[test] - fn test_system_time_to_timestamp() { + fn system_time_to_timestamp() { let now = SystemTime::now(); - let ts = system_time_to_timestamp(now); + let ts = super::system_time_to_timestamp(now); assert!(ts.seconds() > 0, "Timestamp seconds should be positive"); } #[test] - fn test_key_values_to_metric_labels() { + fn key_values_to_metric_labels() { let attrs = [ opentelemetry::KeyValue::new("method", "ExecuteSql"), opentelemetry::KeyValue::new("status.code", "OK"), @@ -413,7 +413,7 @@ mod tests { opentelemetry::KeyValue::new("is_retry", true), opentelemetry::KeyValue::new("instance_id", "my-instance"), ]; - let labels = key_values_to_metric_labels(attrs.iter()); + let labels = super::key_values_to_metric_labels(attrs.iter()); assert_eq!(labels.get("method").map(|s| s.as_str()), Some("ExecuteSql")); assert_eq!(labels.get("status_code").map(|s| s.as_str()), Some("OK")); assert_eq!(labels.get("retry_count").map(|s| s.as_str()), Some("3")); @@ -422,7 +422,7 @@ mod tests { } #[test] - fn test_resource_to_monitored_resource_filtering() { + fn resource_to_monitored_resource_filtering() { let resource = Resource::builder() .with_attributes([ opentelemetry::KeyValue::new("project_id", "my-project"), @@ -435,7 +435,7 @@ mod tests { ]) .build(); - let monitored_res = resource_to_monitored_resource(&resource); + let monitored_res = super::resource_to_monitored_resource(&resource); assert_eq!(monitored_res.r#type, "spanner_instance_client"); assert_eq!( @@ -467,15 +467,15 @@ mod tests { } #[test] - fn test_create_time_series() { + fn create_time_series() { let now = SystemTime::now(); let attrs = [opentelemetry::KeyValue::new("method", "Commit")]; let typed_val = TypedValue::new().set_value(Value::Int64Value(42)); let resource = Resource::builder() .with_attributes([opentelemetry::KeyValue::new("instance_id", "test-instance")]) .build(); - let monitored_resource = resource_to_monitored_resource(&resource); - let ts = create_time_series( + let monitored_resource = super::resource_to_monitored_resource(&resource); + let ts = super::create_time_series( "spanner.googleapis.com/internal/client/operation_count", &monitored_resource, attrs.iter(), @@ -510,7 +510,7 @@ mod tests { } #[test] - fn test_convert_metric_to_time_series_histogram_and_sums() { + fn convert_metric_to_time_series_histogram_and_sums() { let exporter = InMemoryMetricExporter::default(); let reader = opentelemetry_sdk::metrics::PeriodicReader::builder(exporter.clone()).build(); let provider = opentelemetry_sdk::metrics::SdkMeterProvider::builder() @@ -542,7 +542,7 @@ mod tests { let mut time_series_list = Vec::new(); for resource_metrics in &resource_metrics_list { - let monitored_res = resource_to_monitored_resource(resource_metrics.resource()); + let monitored_res = super::resource_to_monitored_resource(resource_metrics.resource()); for scope_metrics in resource_metrics.scope_metrics() { for m in scope_metrics.metrics() { convert_metric_to_time_series(m, &monitored_res, &mut time_series_list); @@ -592,7 +592,7 @@ mod tests { } #[test] - fn test_resource_metrics_scope_filtering() { + fn resource_metrics_scope_filtering() { let exporter = InMemoryMetricExporter::default(); let reader = opentelemetry_sdk::metrics::PeriodicReader::builder(exporter.clone()).build(); @@ -612,7 +612,7 @@ mod tests { let mut time_series_list = Vec::new(); for resource_metrics in &resource_metrics_list { - let monitored_res = resource_to_monitored_resource(resource_metrics.resource()); + let monitored_res = super::resource_to_monitored_resource(resource_metrics.resource()); for scope_metrics in resource_metrics.scope_metrics() { let scope_name = scope_metrics.scope().name(); if scope_name != SPANNER_METER_NAME @@ -636,20 +636,20 @@ mod tests { } #[test] - fn test_is_permission_denied() { + fn is_permission_denied() { let status_pd = google_cloud_gax::error::rpc::Status::default() .set_code(google_cloud_gax::error::rpc::Code::PermissionDenied); let err_pd = crate::Error::service(status_pd); - assert!(is_permission_denied(&err_pd)); + assert!(super::is_permission_denied(&err_pd)); let status_nf = google_cloud_gax::error::rpc::Status::default() .set_code(google_cloud_gax::error::rpc::Code::NotFound); let err_nf = crate::Error::service(status_nf); - assert!(!is_permission_denied(&err_nf)); + assert!(!super::is_permission_denied(&err_nf)); } #[test] - fn test_value_to_string_all_variants() { + fn value_to_string_all_variants() { assert_eq!( value_to_string(&opentelemetry::Value::from("hello")), "hello" diff --git a/src/spanner/src/observability/metrics.rs b/src/spanner/src/observability/metrics.rs index 4856f0d0aa..9fc36ab78f 100644 --- a/src/spanner/src/observability/metrics.rs +++ b/src/spanner/src/observability/metrics.rs @@ -14,22 +14,28 @@ use crate::omni::InstanceType; use gaxi::options::ClientConfig; -use std::time::Duration; - -#[cfg(feature = "_experimental-builtin-metrics")] +use google_cloud_gax::error::Error; +use http::HeaderMap; +use std::fmt::Debug; +use std::future::Future; use std::sync::Arc; -#[cfg(feature = "_experimental-builtin-metrics")] -use std::time::Instant; +use std::time::Duration; #[cfg(feature = "_experimental-builtin-metrics")] use { crate::observability::exporter::GcpMonitoringExporter, + gaxi::attempt_interceptor::AttemptInterceptor, + google_cloud_gax::options::RequestOptions, google_cloud_monitoring_v3::client::MetricService, + http::header::{HeaderName, HeaderValue}, opentelemetry::metrics::{Counter, Histogram, Meter, MeterProvider}, opentelemetry_sdk::{ error::OTelSdkError, metrics::{PeriodicReader, SdkMeterProvider}, }, + std::borrow::Cow, + std::sync::LazyLock, + std::time::Instant, }; #[cfg(feature = "_experimental-builtin-metrics")] @@ -44,7 +50,6 @@ pub(crate) const BUCKET_BOUNDARIES: [f64; 50] = [ ]; #[cfg(feature = "_experimental-builtin-metrics")] -#[allow(dead_code)] #[derive(Debug)] pub(crate) struct SpannerMetrics { pub(crate) operation_latencies: Histogram, @@ -53,6 +58,9 @@ pub(crate) struct SpannerMetrics { pub(crate) afe_latencies: Histogram, pub(crate) operation_count: Counter, pub(crate) attempt_count: Counter, + pub(crate) gfe_connectivity_error_count: Counter, + #[allow(dead_code)] + pub(crate) afe_connectivity_error_count: Counter, } #[cfg(feature = "_experimental-builtin-metrics")] @@ -85,6 +93,12 @@ impl SpannerMetrics { attempt_count: meter .u64_counter("spanner.googleapis.com/internal/client/attempt_count") .build(), + gfe_connectivity_error_count: meter + .u64_counter("spanner.googleapis.com/internal/client/gfe_connectivity_error_count") + .build(), + afe_connectivity_error_count: meter + .u64_counter("spanner.googleapis.com/internal/client/afe_connectivity_error_count") + .build(), } } } @@ -92,7 +106,6 @@ impl SpannerMetrics { /// Parses `projects/{project}/instances/{instance}/databases/{database}` into its /// `(project_id, instance_id, database_id)` components. #[cfg(feature = "_experimental-builtin-metrics")] -#[allow(dead_code)] pub(crate) fn parse_database_name(database_name: &str) -> Option<(&str, &str, &str)> { let mut parts = database_name.split('/'); if parts.next() != Some("projects") { @@ -116,7 +129,6 @@ pub(crate) fn parse_database_name(database_name: &str) -> Option<(&str, &str, &s /// Generates a unique identifier for the `client_uid` metric attribute in the format /// `UUID@PID@hostname`. #[cfg(feature = "_experimental-builtin-metrics")] -#[allow(dead_code)] pub(crate) fn generate_client_uid() -> String { let uuid = uuid::Uuid::new_v4().to_string(); let pid = std::process::id(); @@ -129,7 +141,6 @@ pub(crate) fn generate_client_uid() -> String { /// Generates a 6-character zero-padded lowercase hexadecimal hash for the `client_hash` /// resource label using the 24 least significant bits of an FNV-1a 64-bit hash of `client_uid`. #[cfg(feature = "_experimental-builtin-metrics")] -#[allow(dead_code)] pub(crate) fn generate_client_hash(client_uid: &str) -> String { if client_uid.is_empty() { return "000000".to_string(); @@ -145,7 +156,6 @@ pub(crate) fn generate_client_hash(client_uid: &str) -> String { /// Returns the library client identification string (`"spanner-rust/"`). #[cfg(feature = "_experimental-builtin-metrics")] -#[allow(dead_code)] pub(crate) fn client_name() -> &'static str { concat!("spanner-rust/", env!("CARGO_PKG_VERSION")) } @@ -172,6 +182,11 @@ impl Observability { } } + #[allow(dead_code)] + pub(crate) fn disabled_arc() -> Arc { + Arc::new(Self::disabled()) + } + pub(crate) async fn init( config: &ClientConfig, instance_type: InstanceType, @@ -181,7 +196,15 @@ impl Observability { let disable_builtin_metrics = std::env::var("SPANNER_DISABLE_BUILTIN_METRICS") .map(|s| s.eq_ignore_ascii_case("true") || s == "1") .unwrap_or(false); - if disable_builtin_metrics || instance_type == InstanceType::Omni || is_emulator { + let is_plaintext = config + .endpoint + .as_ref() + .is_some_and(|ep| crate::omni::is_plaintext_endpoint(ep)); + if disable_builtin_metrics + || instance_type == InstanceType::Omni + || is_emulator + || is_plaintext + { return Self::disabled(); } @@ -196,16 +219,16 @@ impl Observability { if let Some(ref cred) = config.cred { builder = builder.with_credentials(cred.clone()); } - if let Some(ref ud) = config.universe_domain { - builder = builder.with_universe_domain(ud.clone()); + if let Some(ref universe_domain) = config.universe_domain { + builder = builder.with_universe_domain(universe_domain.clone()); } let monitoring_client = match builder.build().await { - Ok(c) => c, - Err(e) => { + Ok(monitoring_client) => monitoring_client, + Err(error) => { tracing::warn!( "Failed to initialize Google Cloud Monitoring client for Spanner metrics: {:?}", - e + error ); return Self::disabled(); } @@ -253,13 +276,27 @@ impl Observability { } } + #[cfg(test)] + pub(crate) fn for_test(metrics: SpannerMetrics, meter_provider: SdkMeterProvider) -> Self { + Self { + metrics: Some(Arc::new(metrics)), + common_attributes: [ + opentelemetry::KeyValue::new("client_uid", "test-uid"), + opentelemetry::KeyValue::new("client_name", "test-name"), + opentelemetry::KeyValue::new("database", "test-db"), + ], + meter_provider: Some(Arc::new(meter_provider)), + } + } + + /// Traces a client operation and records operation metrics. pub(crate) async fn trace_operation( &self, method: &'static str, fut: Fut, ) -> crate::Result where - Fut: std::future::Future>, + Fut: Future>, { if self.metrics.is_none() { return fut.await; @@ -271,84 +308,73 @@ impl Observability { result } - #[allow(dead_code)] - pub(crate) async fn trace_attempt( - &self, - method: &'static str, - f: F, - ) -> crate::Result - where - F: FnOnce() -> Fut, - Fut: std::future::Future>, - { - let start_time = Instant::now(); - let result = f().await; - let elapsed = start_time.elapsed(); - self.record_attempt(method, elapsed, &result, None, None); - result - } - - #[allow(dead_code)] - pub(crate) fn record_attempt( + pub(crate) fn record_operation( &self, method: &'static str, duration: Duration, result: &crate::Result, - gfe_latency: Option, - afe_latency: Option, ) { let Some(ref metrics) = self.metrics else { return; }; let status = result_to_status_str(result); + let method_name = normalize_method_name(method); let attributes = [ - opentelemetry::KeyValue::new("method", method), + opentelemetry::KeyValue::new("method", method_name), opentelemetry::KeyValue::new("status", status), opentelemetry::KeyValue::new("directpath_enabled", "false"), - opentelemetry::KeyValue::new("directpath_used", "false"), self.common_attributes[0].clone(), self.common_attributes[1].clone(), self.common_attributes[2].clone(), ]; metrics - .attempt_latencies + .operation_latencies .record(duration.as_secs_f64() * 1000.0, &attributes); - metrics.attempt_count.add(1, &attributes); - - if let Some(gfe) = gfe_latency { - metrics.gfe_latencies.record(gfe, &attributes); - } - if let Some(afe) = afe_latency { - metrics.afe_latencies.record(afe, &attributes); - } + metrics.operation_count.add(1, &attributes); } - pub(crate) fn record_operation( + /// Records metrics for a single RPC attempt, including attempt latency, attempt count, + /// and server-timing metrics (GFE and AFE latency / connectivity errors) extracted from headers. + pub(crate) fn record_attempt( &self, - method: &'static str, + method: &str, duration: Duration, - result: &crate::Result, + error: Option<&Error>, + headers: Option<&HeaderMap>, ) { let Some(ref metrics) = self.metrics else { return; }; - let status = result_to_status_str(result); + let timings = headers.map_or_else(ServerTimings::default, parse_server_timing_from_headers); + let status = error.map_or("OK", error_to_status_str); + let method_name = normalize_method_name(method); let attributes = [ - opentelemetry::KeyValue::new("method", method), + opentelemetry::KeyValue::new("method", method_name), opentelemetry::KeyValue::new("status", status), opentelemetry::KeyValue::new("directpath_enabled", "false"), + opentelemetry::KeyValue::new("directpath_used", "false"), self.common_attributes[0].clone(), self.common_attributes[1].clone(), self.common_attributes[2].clone(), ]; metrics - .operation_latencies + .attempt_latencies .record(duration.as_secs_f64() * 1000.0, &attributes); - metrics.operation_count.add(1, &attributes); + metrics.attempt_count.add(1, &attributes); + + // DirectPath is not used; record GFE latency or connectivity error counter + if let Some(gfe) = timings.gfe_latency { + metrics.gfe_latencies.record(gfe, &attributes); + } else { + metrics.gfe_connectivity_error_count.add(1, &attributes); + } + if let Some(afe) = timings.afe_latency { + metrics.afe_latencies.record(afe, &attributes); + } } pub(crate) fn shutdown(&self) { @@ -371,11 +397,124 @@ impl Drop for Observability { } } +#[cfg(feature = "_experimental-builtin-metrics")] +pub(crate) const AFE_SERVER_TIMING_HEADER: &str = "x-goog-spanner-enable-afe-server-timing"; + +#[cfg(feature = "_experimental-builtin-metrics")] +static AFE_SERVER_TIMING_ENABLED: LazyLock = LazyLock::new(|| { + !std::env::var("SPANNER_DISABLE_AFE_SERVER_TIMING") + .map(|val| val.eq_ignore_ascii_case("true") || val == "1") + .unwrap_or(false) +}); + +#[cfg(feature = "_experimental-builtin-metrics")] +#[inline] +fn is_afe_server_timing_enabled() -> bool { + *AFE_SERVER_TIMING_ENABLED +} + +#[cfg(feature = "_experimental-builtin-metrics")] +pub(crate) fn normalize_method_name(method: &str) -> Cow<'static, str> { + let trimmed = method.trim_start_matches('/'); + let clean = if let Some(suffix) = trimmed.strip_prefix("google.spanner.v1.") { + suffix + } else if let Some(suffix) = trimmed.strip_prefix("Spanner.") { + suffix + } else { + trimmed + }; + + match clean { + "CreateSession" | "Spanner/CreateSession" => Cow::Borrowed("Spanner.CreateSession"), + "BatchCreateSessions" | "Spanner/BatchCreateSessions" => { + Cow::Borrowed("Spanner.BatchCreateSessions") + } + "GetSession" | "Spanner/GetSession" => Cow::Borrowed("Spanner.GetSession"), + "ListSessions" | "Spanner/ListSessions" => Cow::Borrowed("Spanner.ListSessions"), + "DeleteSession" | "Spanner/DeleteSession" => Cow::Borrowed("Spanner.DeleteSession"), + "ExecuteSql" | "Spanner/ExecuteSql" => Cow::Borrowed("Spanner.ExecuteSql"), + "ExecuteStreamingSql" | "Spanner/ExecuteStreamingSql" => { + Cow::Borrowed("Spanner.ExecuteStreamingSql") + } + "ExecuteBatchDml" | "Spanner/ExecuteBatchDml" => Cow::Borrowed("Spanner.ExecuteBatchDml"), + "Read" | "Spanner/Read" => Cow::Borrowed("Spanner.Read"), + "StreamingRead" | "Spanner/StreamingRead" => Cow::Borrowed("Spanner.StreamingRead"), + "BeginTransaction" | "Spanner/BeginTransaction" => { + Cow::Borrowed("Spanner.BeginTransaction") + } + "Commit" | "Spanner/Commit" => Cow::Borrowed("Spanner.Commit"), + "Rollback" | "Spanner/Rollback" => Cow::Borrowed("Spanner.Rollback"), + "PartitionQuery" | "Spanner/PartitionQuery" => Cow::Borrowed("Spanner.PartitionQuery"), + "PartitionRead" | "Spanner/PartitionRead" => Cow::Borrowed("Spanner.PartitionRead"), + "BatchWrite" | "Spanner/BatchWrite" => Cow::Borrowed("Spanner.BatchWrite"), + _ => { + if let Some(suffix) = clean.strip_prefix("Spanner/") { + Cow::Owned(format!("Spanner.{}", suffix.replace('/', "."))) + } else { + Cow::Owned(format!("Spanner.{}", clean.replace('/', "."))) + } + } + } +} + +#[cfg(feature = "_experimental-builtin-metrics")] +fn error_to_status_str(error: &Error) -> &'static str { + error + .status() + .map_or("UNKNOWN", |status| status.code.name()) +} + #[cfg(feature = "_experimental-builtin-metrics")] fn result_to_status_str(result: &crate::Result) -> &'static str { match result { Ok(_) => "OK", - Err(e) => e.status().map_or("UNKNOWN", |status| status.code.name()), + Err(error) => error_to_status_str(error), + } +} + +#[cfg(feature = "_experimental-builtin-metrics")] +pub(crate) fn parse_server_timing_from_headers(headers: &HeaderMap) -> ServerTimings { + let mut timings = ServerTimings::default(); + for header_value in headers.get_all("server-timing") { + let Ok(header_str) = header_value.to_str() else { + continue; + }; + let parsed = parse_server_timing(header_str); + timings.gfe_latency = timings.gfe_latency.or(parsed.gfe_latency); + timings.afe_latency = timings.afe_latency.or(parsed.afe_latency); + } + timings +} + +#[derive(Debug, Default, Clone)] +#[cfg(feature = "_experimental-builtin-metrics")] +pub(crate) struct SpannerMetricsInterceptor; + +#[cfg(feature = "_experimental-builtin-metrics")] +impl AttemptInterceptor for SpannerMetricsInterceptor { + fn intercept(&self, headers: &mut HeaderMap, _attempt: u32) { + if is_afe_server_timing_enabled() { + headers.insert( + HeaderName::from_static(AFE_SERVER_TIMING_HEADER), + HeaderValue::from_static("true"), + ); + } + } + + fn on_attempt_complete( + &self, + method: &str, + _attempt: u32, + start_time: Instant, + response_headers: Option<&HeaderMap>, + error: Option<&Error>, + options: &RequestOptions, + ) { + use google_cloud_gax::options::internal::RequestOptionsExt as _; + if let Some(o11y) = options.get_extension::>() { + let duration = start_time.elapsed(); + o11y.record_attempt(method, duration, error, response_headers); + } } } @@ -387,7 +526,6 @@ pub(crate) struct ServerTimings { } #[cfg(feature = "_experimental-builtin-metrics")] -#[allow(dead_code)] pub(crate) fn parse_server_timing(header_val: &str) -> ServerTimings { let mut timings = ServerTimings::default(); for part in header_val.split(',') { @@ -402,10 +540,10 @@ pub(crate) fn parse_server_timing(header_val: &str) -> ServerTimings { } if let Some(duration) = subparts.find_map(parse_duration_param) { if is_gfe { - timings.gfe_latency = Some(duration); + timings.gfe_latency = timings.gfe_latency.or(Some(duration)); } if is_afe { - timings.afe_latency = Some(duration); + timings.afe_latency = timings.afe_latency.or(Some(duration)); } } } @@ -437,6 +575,11 @@ impl Observability { Self } + #[allow(dead_code)] + pub(crate) fn disabled_arc() -> Arc { + Arc::new(Self::disabled()) + } + pub(crate) async fn init( _config: &ClientConfig, _instance_type: InstanceType, @@ -453,39 +596,23 @@ impl Observability { fut: Fut, ) -> crate::Result where - Fut: std::future::Future>, + Fut: Future>, { fut.await } - /// No-op stub implementation when the `_experimental-builtin-metrics` feature is disabled. - #[inline(always)] - #[allow(dead_code)] - pub(crate) async fn trace_attempt( - &self, - _method: &'static str, - f: F, - ) -> crate::Result - where - F: FnOnce() -> Fut, - Fut: std::future::Future>, - { - f().await - } - /// No-op stub implementation when the `_experimental-builtin-metrics` feature is disabled. /// - /// This allows client operations to call `record_attempt` unconditionally without sprinkling + /// This allows interceptors to call `record_attempt` unconditionally without sprinkling /// `#[cfg(feature = "_experimental-builtin-metrics")]` across call sites. #[inline(always)] #[allow(dead_code)] - pub(crate) fn record_attempt( + pub(crate) fn record_attempt( &self, - _method: &'static str, + _method: &str, _duration: Duration, - _result: &crate::Result, - _gfe_latency: Option, - _afe_latency: Option, + _error: Option<&Error>, + _headers: Option<&HeaderMap>, ) { } @@ -507,6 +634,33 @@ impl Observability { pub(crate) fn shutdown(&self) {} } +#[cfg(all(test, not(feature = "_experimental-builtin-metrics")))] +mod disabled_tests { + use super::*; + + #[tokio::test] + async fn disabled_stubs_exercise() { + let o11y = Observability::disabled(); + let _o11y_arc = Observability::disabled_arc(); + let initialized = Observability::init( + &ClientConfig::default(), + InstanceType::Cloud, + "projects/p/instances/i/databases/d", + false, + ) + .await; + initialized.record_attempt("ExecuteSql", Duration::from_millis(10), None, None); + let ok_res: crate::Result<()> = Ok(()); + initialized.record_operation("ExecuteSql", Duration::from_millis(10), &ok_res); + let res = initialized + .trace_operation("ExecuteSql", async { Ok::<_, crate::Error>(42) }) + .await + .expect("trace_operation should succeed"); + assert_eq!(res, 42); + o11y.shutdown(); + } +} + #[cfg(all(test, feature = "_experimental-builtin-metrics"))] mod tests { use super::*; @@ -519,12 +673,115 @@ mod tests { fn traits() { static_assertions::assert_impl_all!(Observability: Send, Sync, Debug, Clone); static_assertions::assert_impl_all!(SpannerMetrics: Send, Sync, Debug); + static_assertions::assert_impl_all!(ServerTimings: Send, Sync, Debug, PartialEq, Default); + static_assertions::assert_impl_all!(SpannerMetricsInterceptor: Send, Sync, Debug, Clone, Default); + } + + #[test] + fn normalize_method_names() { + assert_eq!( + normalize_method_name("/google.spanner.v1.Spanner/CreateSession"), + "Spanner.CreateSession" + ); + assert_eq!( + normalize_method_name("google.spanner.v1.Spanner/BatchCreateSessions"), + "Spanner.BatchCreateSessions" + ); + assert_eq!( + normalize_method_name("google.spanner.v1.Spanner/ExecuteBatchDml"), + "Spanner.ExecuteBatchDml" + ); + assert_eq!( + normalize_method_name("Spanner.BeginTransaction"), + "Spanner.BeginTransaction" + ); + assert_eq!(normalize_method_name("Commit"), "Spanner.Commit"); + assert_eq!(normalize_method_name("Rollback"), "Spanner.Rollback"); + assert_eq!( + normalize_method_name("PartitionQuery"), + "Spanner.PartitionQuery" + ); + assert_eq!( + normalize_method_name("PartitionRead"), + "Spanner.PartitionRead" + ); + assert_eq!(normalize_method_name("BatchWrite"), "Spanner.BatchWrite"); + assert_eq!( + normalize_method_name("ExecuteStreamingSql"), + "Spanner.ExecuteStreamingSql" + ); + assert_eq!(normalize_method_name("Read"), "Spanner.Read"); + assert_eq!( + normalize_method_name("StreamingRead"), + "Spanner.StreamingRead" + ); + assert_eq!( + normalize_method_name("DeleteSession"), + "Spanner.DeleteSession" + ); + assert_eq!(normalize_method_name("GetSession"), "Spanner.GetSession"); + assert_eq!( + normalize_method_name("ListSessions"), + "Spanner.ListSessions" + ); + assert_eq!( + normalize_method_name("CustomOperation"), + "Spanner.CustomOperation" + ); + assert_eq!( + normalize_method_name("Spanner/CustomOp"), + "Spanner.CustomOp" + ); + assert_eq!( + normalize_method_name("Spanner/Sub/Method"), + "Spanner.Sub.Method" + ); + assert_eq!( + normalize_method_name("/google.spanner.v1.Spanner/CustomOp"), + "Spanner.CustomOp" + ); + assert_eq!( + normalize_method_name("google.spanner.v1.Spanner/CustomOp"), + "Spanner.CustomOp" + ); + assert_eq!( + normalize_method_name("Spanner.CustomOp"), + "Spanner.CustomOp" + ); + assert_eq!( + normalize_method_name("Spanner.Sub/Method"), + "Spanner.Sub.Method" + ); + assert_eq!( + normalize_method_name("google.spanner.v1.Spanner/Sub/Method"), + "Spanner.Sub.Method" + ); } #[test] fn observability_disabled() { let o11y = Observability::disabled(); assert!(o11y.metrics.is_none()); + let o11y_arc = Observability::disabled_arc(); + assert!(o11y_arc.metrics.is_none()); + + let ok_res: crate::Result<()> = Ok(()); + o11y.record_operation("ExecuteSql", Duration::from_millis(10), &ok_res); + o11y.record_attempt("ExecuteSql", Duration::from_millis(10), None, None); + o11y.shutdown(); + } + + #[test] + fn spanner_metrics_interceptor_intercept_sets_header() { + let interceptor = SpannerMetricsInterceptor; + let mut headers = HeaderMap::new(); + interceptor.intercept(&mut headers, 1); + assert_eq!( + headers + .get(AFE_SERVER_TIMING_HEADER) + .map(|v| v.to_str().expect("valid ascii")), + Some("true") + ); } #[test] @@ -536,6 +793,14 @@ mod tests { .set_code(google_cloud_gax::error::rpc::Code::PermissionDenied); let err_pd: crate::Result<()> = Err(crate::Error::service(status_pd)); assert_eq!(result_to_status_str(&err_pd), "PERMISSION_DENIED"); + + let status_nf = google_cloud_gax::error::rpc::Status::default() + .set_code(google_cloud_gax::error::rpc::Code::NotFound); + let err_nf: crate::Result<()> = Err(crate::Error::service(status_nf)); + assert_eq!(result_to_status_str(&err_nf), "NOT_FOUND"); + + let err_other: crate::Result<()> = Err(crate::Error::timeout("some generic timeout")); + assert_eq!(result_to_status_str(&err_other), "UNKNOWN"); } #[test] @@ -557,12 +822,16 @@ mod tests { let ok_res: crate::Result<()> = Ok(()); o11y.record_operation("ExecuteSql", Duration::from_millis(50), &ok_res); + let mut headers = HeaderMap::new(); + headers.insert( + "server-timing", + HeaderValue::from_static("gfet4t7;dur=12.5,afe;dur=5.0"), + ); o11y.record_attempt( "ExecuteSql", Duration::from_millis(40), - &ok_res, - Some(12.5), - Some(5.0), + None, + Some(&headers), ); provider.force_flush().expect("force_flush failed"); @@ -735,7 +1004,7 @@ mod tests { assert_eq!( super::parse_server_timing("gfet4t7;dur=10.0, gfet4t7;dur=20.0"), ServerTimings { - gfe_latency: Some(20.0), + gfe_latency: Some(10.0), afe_latency: None, } ); @@ -748,6 +1017,139 @@ mod tests { ); } + #[test] + fn parse_server_timing_from_headers_multiple_headers() { + let mut headers = HeaderMap::new(); + headers.append( + "server-timing", + http::HeaderValue::from_static("gfet4t7;dur=15.5"), + ); + headers.append( + "server-timing", + http::HeaderValue::from_static("afe;dur=7.2"), + ); + + let timings = parse_server_timing_from_headers(&headers); + assert_eq!(timings.gfe_latency, Some(15.5)); + assert_eq!(timings.afe_latency, Some(7.2)); + + let mut invalid_headers = HeaderMap::new(); + invalid_headers.append( + "server-timing", + http::HeaderValue::from_bytes(b"\xff\xfe").expect("valid raw header bytes"), + ); + let invalid_timings = parse_server_timing_from_headers(&invalid_headers); + assert_eq!(invalid_timings, ServerTimings::default()); + } + + #[tokio::test] + async fn trace_operation_records_operation_metrics() { + let exporter = InMemoryMetricExporter::default(); + let reader = PeriodicReader::builder(exporter.clone()).build(); + let provider = SdkMeterProvider::builder().with_reader(reader).build(); + let meter = provider.meter("cloud.google.com/rust"); + let metrics = SpannerMetrics::new(meter); + let o11y = Observability { + metrics: Some(Arc::new(metrics)), + common_attributes: [ + opentelemetry::KeyValue::new("client_uid", "test-uid"), + opentelemetry::KeyValue::new("client_name", "test-name"), + opentelemetry::KeyValue::new("database", "test-db"), + ], + meter_provider: Some(Arc::new(provider.clone())), + }; + + let result = o11y + .trace_operation("ExecuteSql", async { Ok::(100) }) + .await; + assert_eq!( + result.expect("trace_operation should succeed"), + 100, + "operation result should match" + ); + + provider.force_flush().expect("force_flush failed"); + + let finished = exporter + .get_finished_metrics() + .expect("get_finished_metrics should succeed"); + let metric_names: Vec<&str> = finished + .iter() + .flat_map(|rm| rm.scope_metrics()) + .flat_map(|sm| sm.metrics()) + .map(|m| m.name()) + .collect(); + + assert!( + metric_names.contains(&"spanner.googleapis.com/internal/client/operation_latencies"), + "should record operation_latencies" + ); + assert!( + metric_names.contains(&"spanner.googleapis.com/internal/client/operation_count"), + "should record operation_count" + ); + } + + #[test] + fn spanner_metrics_interceptor_records_attempt_metrics() { + use gaxi::attempt_interceptor::AttemptInterceptor; + use google_cloud_gax::options::internal::RequestOptionsExt as _; + + let exporter = InMemoryMetricExporter::default(); + let reader = PeriodicReader::builder(exporter.clone()).build(); + let provider = SdkMeterProvider::builder().with_reader(reader).build(); + let meter = provider.meter("cloud.google.com/rust"); + let metrics = SpannerMetrics::new(meter); + let o11y = Arc::new(Observability::for_test(metrics, provider.clone())); + let interceptor = SpannerMetricsInterceptor; + + let mut res_headers = HeaderMap::new(); + res_headers.insert( + "server-timing", + http::HeaderValue::from_static("gfet4t7;dur=12.5,afe;dur=3.2"), + ); + let options = crate::RequestOptions::default().insert_extension(o11y); + let start_time = Instant::now(); + + interceptor.on_attempt_complete( + "/google.spanner.v1.Spanner/ExecuteSql", + 1, + start_time, + Some(&res_headers), + None, + &options, + ); + + provider.force_flush().expect("force_flush failed"); + + let finished = exporter + .get_finished_metrics() + .expect("get_finished_metrics should succeed"); + let metric_names: Vec<&str> = finished + .iter() + .flat_map(|rm| rm.scope_metrics()) + .flat_map(|sm| sm.metrics()) + .map(|m| m.name()) + .collect(); + + assert!( + metric_names.contains(&"spanner.googleapis.com/internal/client/attempt_latencies"), + "should record attempt_latencies" + ); + assert!( + metric_names.contains(&"spanner.googleapis.com/internal/client/attempt_count"), + "should record attempt_count" + ); + assert!( + metric_names.contains(&"spanner.googleapis.com/internal/client/gfe_latencies"), + "should record gfe_latencies" + ); + assert!( + metric_names.contains(&"spanner.googleapis.com/internal/client/afe_latencies"), + "should record afe_latencies" + ); + } + #[test] fn default_export_interval() { assert_eq!(DEFAULT_EXPORT_INTERVAL, Duration::from_secs(60)); @@ -797,6 +1199,27 @@ mod tests { o11y_omni.metrics.is_none(), "omni client should have disabled metrics" ); + + let mut plaintext_config = ClientConfig::default(); + plaintext_config.endpoint = Some("http://localhost:9010".to_string()); + let o11y_plaintext = Observability::init( + &plaintext_config, + InstanceType::Cloud, + "projects/proj/instances/inst/databases/db", + false, + ) + .await; + assert!( + o11y_plaintext.metrics.is_none(), + "plaintext endpoint must disable metrics" + ); + + let o11y_invalid_db = + Observability::init(&config, InstanceType::Cloud, "invalid-db-name", false).await; + assert!( + o11y_invalid_db.metrics.is_none(), + "invalid database name must return disabled observability" + ); } #[test] @@ -834,8 +1257,8 @@ mod tests { let mut metric_names = Vec::new(); for resource_metrics in &finished_metrics { for scope_metrics in resource_metrics.scope_metrics() { - for m in scope_metrics.metrics() { - metric_names.push(m.name().to_string()); + for metric in scope_metrics.metrics() { + metric_names.push(metric.name().to_string()); } } } @@ -915,13 +1338,12 @@ mod tests { }; o11y.record_operation("test_op", Duration::from_millis(15), &Ok(())); - o11y.record_attempt( - "test_op", - Duration::from_millis(10), - &Ok(()), - Some(3.0), - Some(2.0), + let mut headers = HeaderMap::new(); + headers.insert( + "server-timing", + HeaderValue::from_static("gfet4t7;dur=3.0,afe;dur=2.0"), ); + o11y.record_attempt("test_op", Duration::from_millis(10), None, Some(&headers)); if let Some(ref provider) = o11y.meter_provider { provider.force_flush().expect("force_flush should succeed"); @@ -942,7 +1364,7 @@ mod tests { .expect("attempt_latencies should be exported"); assert_eq!( attempt_attrs.get("method").map(String::as_str), - Some("test_op") + Some("Spanner.test_op") ); assert_eq!(attempt_attrs.get("status").map(String::as_str), Some("OK")); assert_eq!( @@ -999,7 +1421,10 @@ mod tests { "spanner.googleapis.com/internal/client/operation_latencies", ) .expect("operation_latencies should be exported"); - assert_eq!(op_attrs.get("method").map(String::as_str), Some("test_op")); + assert_eq!( + op_attrs.get("method").map(String::as_str), + Some("Spanner.test_op") + ); assert_eq!(op_attrs.get("status").map(String::as_str), Some("OK")); assert_eq!( op_attrs.get("directpath_enabled").map(String::as_str), @@ -1030,20 +1455,265 @@ mod tests { #[cfg(feature = "_experimental-builtin-metrics")] #[test] - fn observability_double_shutdown_ignores_already_shutdown() { - let meter_provider = SdkMeterProvider::builder().build(); - let o11y = Observability { - metrics: None, - common_attributes: [ - opentelemetry::KeyValue::new("client_uid", ""), - opentelemetry::KeyValue::new("client_name", ""), - opentelemetry::KeyValue::new("database", ""), - ], - meter_provider: Some(Arc::new(meter_provider)), - }; - // Calling shutdown twice should cleanly handle AlreadyShutdown on the second call. - o11y.shutdown(); - o11y.shutdown(); - // Dropping o11y invokes Drop and calls shutdown a third time without panic or warning. + fn spanner_metrics_interceptor_traits() { + static_assertions::assert_impl_all!(SpannerMetricsInterceptor: Send, Sync, Debug, Clone); + } + + #[test] + fn afe_server_timing_request_header_sent() { + let interceptor = SpannerMetricsInterceptor; + let mut headers = HeaderMap::new(); + interceptor.intercept(&mut headers, 1); + + assert_eq!( + headers + .get(AFE_SERVER_TIMING_HEADER) + .and_then(|value| value.to_str().ok()), + Some("true"), + "should add x-goog-spanner-enable-afe-server-timing header by default" + ); + } + + #[test] + fn multi_attempt_retry_metrics() { + use google_cloud_gax::options::internal::RequestOptionsExt as _; + + let exporter = InMemoryMetricExporter::default(); + let reader = PeriodicReader::builder(exporter.clone()).build(); + let provider = SdkMeterProvider::builder().with_reader(reader).build(); + let meter = provider.meter("cloud.google.com/rust"); + let metrics = SpannerMetrics::new(meter); + let o11y = Arc::new(Observability::for_test(metrics, provider.clone())); + let interceptor = SpannerMetricsInterceptor; + + let options = crate::RequestOptions::default().insert_extension(o11y.clone()); + let start_time = Instant::now(); + + // Attempt 1 fails with UNAVAILABLE + let error_unavailable = Error::service( + google_cloud_gax::error::rpc::Status::default() + .set_code(google_cloud_gax::error::rpc::Code::Unavailable) + .set_message("service unavailable"), + ); + interceptor.on_attempt_complete( + "/google.spanner.v1.Spanner/ExecuteSql", + 1, + start_time, + None, + Some(&error_unavailable), + &options, + ); + + // Attempt 2 succeeds + let mut res_headers = HeaderMap::new(); + res_headers.insert( + "server-timing", + HeaderValue::from_static("gfet4t7;dur=10.0,afe;dur=2.0"), + ); + interceptor.on_attempt_complete( + "/google.spanner.v1.Spanner/ExecuteSql", + 2, + start_time, + Some(&res_headers), + None, + &options, + ); + + // Record total operation completion + o11y.record_operation( + "google.spanner.v1.Spanner/ExecuteSql", + Duration::from_millis(50), + &Ok::<(), crate::Error>(()), + ); + + provider.force_flush().expect("force_flush should succeed"); + let finished = exporter + .get_finished_metrics() + .expect("get_finished_metrics should succeed"); + + let attempt_count_attrs = extract_all_attributes( + &finished, + "spanner.googleapis.com/internal/client/attempt_count", + ); + assert_eq!(attempt_count_attrs.len(), 2, "should record 2 attempts"); + let statuses: Vec<&str> = attempt_count_attrs + .iter() + .map(|attribute_map| { + attribute_map + .get("status") + .map(String::as_str) + .unwrap_or("") + }) + .collect(); + assert!( + statuses.contains(&"UNAVAILABLE"), + "should contain UNAVAILABLE attempt" + ); + assert!(statuses.contains(&"OK"), "should contain OK attempt"); + + for attr in &attempt_count_attrs { + assert_eq!( + attr.get("method").map(String::as_str), + Some("Spanner.ExecuteSql"), + "method attribute should be normalized across attempts" + ); + } + + let op_count_attrs = extract_all_attributes( + &finished, + "spanner.googleapis.com/internal/client/operation_count", + ); + assert_eq!(op_count_attrs.len(), 1, "should record 1 operation"); + assert_eq!( + op_count_attrs[0].get("method").map(String::as_str), + Some("Spanner.ExecuteSql"), + "operation method attribute should be normalized" + ); + assert_eq!( + op_count_attrs[0].get("status").map(String::as_str), + Some("OK") + ); + } + + #[test] + fn transport_error_status_unknown() { + use google_cloud_gax::options::internal::RequestOptionsExt as _; + + let exporter = InMemoryMetricExporter::default(); + let reader = PeriodicReader::builder(exporter.clone()).build(); + let provider = SdkMeterProvider::builder().with_reader(reader).build(); + let meter = provider.meter("cloud.google.com/rust"); + let metrics = SpannerMetrics::new(meter); + let o11y = Arc::new(Observability::for_test(metrics, provider.clone())); + let interceptor = SpannerMetricsInterceptor; + + let options = crate::RequestOptions::default().insert_extension(o11y); + let start_time = Instant::now(); + + // Non-gRPC transport error (e.g. IO failure without gRPC Status) + let transport_error = Error::timeout("simulated timeout"); + interceptor.on_attempt_complete( + "/google.spanner.v1.Spanner/ExecuteSql", + 1, + start_time, + None, + Some(&transport_error), + &options, + ); + + provider.force_flush().expect("force_flush should succeed"); + let finished = exporter + .get_finished_metrics() + .expect("get_finished_metrics should succeed"); + + let attempt_attrs = extract_all_attributes( + &finished, + "spanner.googleapis.com/internal/client/attempt_count", + ); + assert_eq!(attempt_attrs.len(), 1); + assert_eq!( + attempt_attrs[0].get("status").map(String::as_str), + Some("UNKNOWN"), + "non-gRPC transport errors must record status = UNKNOWN" + ); + } + + #[test] + fn missing_server_timing_increments_gfe_connectivity_error_counter() { + use google_cloud_gax::options::internal::RequestOptionsExt as _; + + let exporter = InMemoryMetricExporter::default(); + let reader = PeriodicReader::builder(exporter.clone()).build(); + let provider = SdkMeterProvider::builder().with_reader(reader).build(); + let meter = provider.meter("cloud.google.com/rust"); + let metrics = SpannerMetrics::new(meter); + let o11y = Arc::new(Observability::for_test(metrics, provider.clone())); + let interceptor = SpannerMetricsInterceptor; + + let options = crate::RequestOptions::default().insert_extension(o11y); + let start_time = Instant::now(); + + // Attempt completes with headers missing server-timing + let empty_headers = HeaderMap::new(); + interceptor.on_attempt_complete( + "/google.spanner.v1.Spanner/CreateSession", + 1, + start_time, + Some(&empty_headers), + None, + &options, + ); + + provider.force_flush().expect("force_flush should succeed"); + let finished = exporter + .get_finished_metrics() + .expect("get_finished_metrics should succeed"); + + let metric_names: Vec<&str> = finished + .iter() + .flat_map(|rm| rm.scope_metrics()) + .flat_map(|sm| sm.metrics()) + .map(|m| m.name()) + .collect(); + + assert!( + metric_names + .contains(&"spanner.googleapis.com/internal/client/gfe_connectivity_error_count"), + "missing server-timing must increment gfe_connectivity_error_count" + ); + assert!( + !metric_names + .contains(&"spanner.googleapis.com/internal/client/afe_connectivity_error_count"), + "non-DirectPath requests must NOT increment afe_connectivity_error_count when AFE timing is absent" + ); + } + + fn extract_all_attributes( + finished: &[ResourceMetrics], + metric_name: &str, + ) -> Vec> { + let mut result = Vec::new(); + for resource_metrics in finished { + for scope_metrics in resource_metrics.scope_metrics() { + for metric in scope_metrics.metrics() { + if metric.name() == metric_name { + match metric.data() { + AggregatedMetrics::U64(MetricData::Sum(sum)) => { + for data_point in sum.data_points() { + result.push( + data_point + .attributes() + .map(|key_value| { + ( + key_value.key.to_string(), + key_value.value.to_string(), + ) + }) + .collect(), + ); + } + } + AggregatedMetrics::F64(MetricData::Histogram(histogram)) => { + for data_point in histogram.data_points() { + result.push( + data_point + .attributes() + .map(|key_value| { + ( + key_value.key.to_string(), + key_value.value.to_string(), + ) + }) + .collect(), + ); + } + } + _ => {} + } + } + } + } + } + result } } diff --git a/src/spanner/src/request_id.rs b/src/spanner/src/request_id.rs index 701bf29e6d..840b4d760b 100644 --- a/src/spanner/src/request_id.rs +++ b/src/spanner/src/request_id.rs @@ -233,7 +233,7 @@ mod tests { request.clone(), crate::RequestOptions::default(), 0, - &crate::observability::Observability::disabled(), + &crate::observability::Observability::disabled_arc(), ) .await .expect("first create_session should succeed after retry"); @@ -244,7 +244,7 @@ mod tests { request, crate::RequestOptions::default(), 0, - &crate::observability::Observability::disabled(), + &crate::observability::Observability::disabled_arc(), ) .await .expect("second create_session should succeed"); diff --git a/src/spanner/src/session_maintainer.rs b/src/spanner/src/session_maintainer.rs index 18fe618890..e75f8a4bf0 100644 --- a/src/spanner/src/session_maintainer.rs +++ b/src/spanner/src/session_maintainer.rs @@ -130,7 +130,7 @@ impl ManagedSessionMaintainer { database_name: &str, database_role: &str, options: &RequestOptions, - o11y: &Observability, + o11y: &Arc, ) -> Result { let request = CreateSessionRequest::new() .set_database(database_name) @@ -562,7 +562,7 @@ mod tests { spanner, session_maintainer: maintainer.clone(), leader_aware_routing_enabled: true, - o11y: std::sync::Arc::new(crate::observability::Observability::disabled()), + o11y: Arc::new(Observability::disabled()), }; // 1. Create builder (captures session 1) From c40404ee2909c70c156d031d9567499504cb84d4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Knut=20Olav=20L=C3=B8ite?= Date: Tue, 11 Aug 2026 08:38:55 +0200 Subject: [PATCH 2/2] chore(spanner): cleanup --- src/spanner/src/client.rs | 20 ++-- src/spanner/src/observability/exporter.rs | 82 ++++++++-------- src/spanner/src/observability/metrics.rs | 72 +++++++------- src/spanner/src/request_id.rs | 110 +++++++++++++++------- 4 files changed, 162 insertions(+), 122 deletions(-) diff --git a/src/spanner/src/client.rs b/src/spanner/src/client.rs index 7ca46301a6..abe406035a 100644 --- a/src/spanner/src/client.rs +++ b/src/spanner/src/client.rs @@ -166,7 +166,7 @@ macro_rules! define_idempotent_rpc { ) -> crate::Result<$response_type> { let options = self.attach_request_id(options, channel_hint); #[cfg(feature = "_experimental-builtin-metrics")] - let options = options.insert_extension(o11y.clone()); + let options = options.insert_extension(Arc::clone(o11y)); o11y.trace_operation( $canonical_name, self.get_channel(channel_hint) @@ -698,7 +698,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), - &crate::observability::Observability::disabled_arc(), + &Observability::disabled_arc(), ) .await .expect("Failed to call create_session"); @@ -818,7 +818,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), - &crate::observability::Observability::disabled_arc(), + &Observability::disabled_arc(), ) .await .expect("Failed to call create_session after transport error retry"); @@ -868,7 +868,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), - &crate::observability::Observability::disabled_arc(), + &Observability::disabled_arc(), ) .await .expect("Failed to call execute_sql"); @@ -912,7 +912,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), - &crate::observability::Observability::disabled_arc(), + &Observability::disabled_arc(), ) .await .expect("Failed to call execute_batch_dml"); @@ -951,7 +951,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), - &crate::observability::Observability::disabled_arc(), + &Observability::disabled_arc(), ) .await .expect("Failed to call begin_transaction"); @@ -994,7 +994,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), - &crate::observability::Observability::disabled_arc(), + &Observability::disabled_arc(), ) .await .expect("Failed to call commit"); @@ -1028,7 +1028,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), - &crate::observability::Observability::disabled_arc(), + &Observability::disabled_arc(), ) .await .expect("Failed to call rollback"); @@ -1261,7 +1261,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), - &crate::observability::Observability::disabled_arc(), + &Observability::disabled_arc(), ) .await .expect("Failed to call create_session"); @@ -1308,7 +1308,7 @@ mod tests { req, options, client.next_channel_hint(), - &crate::observability::Observability::disabled_arc(), + &Observability::disabled_arc(), ) .await; diff --git a/src/spanner/src/observability/exporter.rs b/src/spanner/src/observability/exporter.rs index 0814e7e576..c59265d98e 100644 --- a/src/spanner/src/observability/exporter.rs +++ b/src/spanner/src/observability/exporter.rs @@ -14,9 +14,11 @@ use google_cloud_api::model::distribution::{BucketOptions, bucket_options::Explicit}; use google_cloud_api::model::{Distribution, Metric, MonitoredResource, metric_descriptor}; +use google_cloud_gax::error::rpc::Code; use google_cloud_monitoring_v3::client::MetricService; use google_cloud_monitoring_v3::model::typed_value::Value; use google_cloud_monitoring_v3::model::{Point, TimeInterval, TimeSeries, TypedValue}; +use opentelemetry::{KeyValue, Value as OTelValue}; use opentelemetry_sdk::Resource; use opentelemetry_sdk::error::{OTelSdkError, OTelSdkResult}; use opentelemetry_sdk::metrics::Temporality; @@ -168,7 +170,7 @@ async fn send_time_series_batches( fn is_permission_denied(err: &crate::Error) -> bool { err.status() - .map(|s| s.code == google_cloud_gax::error::rpc::Code::PermissionDenied) + .map(|s| s.code == Code::PermissionDenied) .unwrap_or(false) } @@ -241,7 +243,7 @@ fn create_time_interval(start: SystemTime, end: SystemTime) -> TimeInterval { .set_end_time(system_time_to_timestamp(end)) } -fn value_to_string(value: &opentelemetry::Value) -> String { +fn value_to_string(value: &OTelValue) -> String { value.to_string() } @@ -253,7 +255,7 @@ fn is_monitored_resource_label(key: &str) -> bool { } fn key_values_to_metric_labels<'a>( - attrs: impl Iterator, + attrs: impl Iterator, ) -> HashMap { let (lower_bound, _) = attrs.size_hint(); let mut labels = HashMap::with_capacity(lower_bound); @@ -292,7 +294,7 @@ fn resource_to_monitored_resource(resource: &Resource) -> MonitoredResource { fn create_time_series<'a>( metric_type: &str, monitored_resource: &MonitoredResource, - attributes: impl Iterator, + attributes: impl Iterator, start_time: SystemTime, end_time: SystemTime, typed_value: TypedValue, @@ -395,8 +397,17 @@ mod tests { use super::*; use opentelemetry::metrics::{Counter, Histogram, MeterProvider as _}; use opentelemetry_sdk::metrics::InMemoryMetricExporter; + use std::fmt::Debug; use std::time::SystemTime; + static_assertions::assert_impl_all!( + GcpMonitoringExporter: Send, + Sync, + Debug, + Clone, + PushMetricExporter + ); + #[test] fn system_time_to_timestamp() { let now = SystemTime::now(); @@ -407,11 +418,11 @@ mod tests { #[test] fn key_values_to_metric_labels() { let attrs = [ - opentelemetry::KeyValue::new("method", "ExecuteSql"), - opentelemetry::KeyValue::new("status.code", "OK"), - opentelemetry::KeyValue::new("retry.count", 3_i64), - opentelemetry::KeyValue::new("is_retry", true), - opentelemetry::KeyValue::new("instance_id", "my-instance"), + KeyValue::new("method", "ExecuteSql"), + KeyValue::new("status.code", "OK"), + KeyValue::new("retry.count", 3_i64), + KeyValue::new("is_retry", true), + KeyValue::new("instance_id", "my-instance"), ]; let labels = super::key_values_to_metric_labels(attrs.iter()); assert_eq!(labels.get("method").map(|s| s.as_str()), Some("ExecuteSql")); @@ -425,13 +436,13 @@ mod tests { fn resource_to_monitored_resource_filtering() { let resource = Resource::builder() .with_attributes([ - opentelemetry::KeyValue::new("project_id", "my-project"), - opentelemetry::KeyValue::new("instance_id", "my-instance"), - opentelemetry::KeyValue::new("location", "us-central1"), - opentelemetry::KeyValue::new("instance_config", "regional-us-central1"), - opentelemetry::KeyValue::new("client_hash", "abc1234"), - opentelemetry::KeyValue::new("service.name", "my-app"), - opentelemetry::KeyValue::new("telemetry.sdk.version", "1.0.0"), + KeyValue::new("project_id", "my-project"), + KeyValue::new("instance_id", "my-instance"), + KeyValue::new("location", "us-central1"), + KeyValue::new("instance_config", "regional-us-central1"), + KeyValue::new("client_hash", "abc1234"), + KeyValue::new("service.name", "my-app"), + KeyValue::new("telemetry.sdk.version", "1.0.0"), ]) .build(); @@ -469,10 +480,10 @@ mod tests { #[test] fn create_time_series() { let now = SystemTime::now(); - let attrs = [opentelemetry::KeyValue::new("method", "Commit")]; + let attrs = [KeyValue::new("method", "Commit")]; let typed_val = TypedValue::new().set_value(Value::Int64Value(42)); let resource = Resource::builder() - .with_attributes([opentelemetry::KeyValue::new("instance_id", "test-instance")]) + .with_attributes([KeyValue::new("instance_id", "test-instance")]) .build(); let monitored_resource = super::resource_to_monitored_resource(&resource); let ts = super::create_time_series( @@ -524,15 +535,9 @@ mod tests { .f64_counter("spanner.googleapis.com/internal/client/custom_latency") .build(); - histogram.record( - 123.45, - &[opentelemetry::KeyValue::new("method", "ExecuteSql")], - ); - counter_u64.add(1, &[opentelemetry::KeyValue::new("method", "ExecuteSql")]); - counter_f64.add( - 99.5, - &[opentelemetry::KeyValue::new("method", "ExecuteSql")], - ); + histogram.record(123.45, &[KeyValue::new("method", "ExecuteSql")]); + counter_u64.add(1, &[KeyValue::new("method", "ExecuteSql")]); + counter_f64.add(99.5, &[KeyValue::new("method", "ExecuteSql")]); provider.force_flush().expect("force_flush failed"); @@ -637,29 +642,22 @@ mod tests { #[test] fn is_permission_denied() { - let status_pd = google_cloud_gax::error::rpc::Status::default() - .set_code(google_cloud_gax::error::rpc::Code::PermissionDenied); + let status_pd = + google_cloud_gax::error::rpc::Status::default().set_code(Code::PermissionDenied); let err_pd = crate::Error::service(status_pd); assert!(super::is_permission_denied(&err_pd)); - let status_nf = google_cloud_gax::error::rpc::Status::default() - .set_code(google_cloud_gax::error::rpc::Code::NotFound); + let status_nf = google_cloud_gax::error::rpc::Status::default().set_code(Code::NotFound); let err_nf = crate::Error::service(status_nf); assert!(!super::is_permission_denied(&err_nf)); } #[test] fn value_to_string_all_variants() { - assert_eq!( - value_to_string(&opentelemetry::Value::from("hello")), - "hello" - ); - assert_eq!(value_to_string(&opentelemetry::Value::from(42_i64)), "42"); - assert_eq!( - value_to_string(&opentelemetry::Value::from(123.456_f64)), - "123.456" - ); - assert_eq!(value_to_string(&opentelemetry::Value::from(true)), "true"); - assert_eq!(value_to_string(&opentelemetry::Value::from(false)), "false"); + assert_eq!(value_to_string(&OTelValue::from("hello")), "hello"); + assert_eq!(value_to_string(&OTelValue::from(42_i64)), "42"); + assert_eq!(value_to_string(&OTelValue::from(123.456_f64)), "123.456"); + assert_eq!(value_to_string(&OTelValue::from(true)), "true"); + assert_eq!(value_to_string(&OTelValue::from(false)), "false"); } } diff --git a/src/spanner/src/observability/metrics.rs b/src/spanner/src/observability/metrics.rs index 9fc36ab78f..9a516c5808 100644 --- a/src/spanner/src/observability/metrics.rs +++ b/src/spanner/src/observability/metrics.rs @@ -28,8 +28,10 @@ use { google_cloud_gax::options::RequestOptions, google_cloud_monitoring_v3::client::MetricService, http::header::{HeaderName, HeaderValue}, + opentelemetry::KeyValue, opentelemetry::metrics::{Counter, Histogram, Meter, MeterProvider}, opentelemetry_sdk::{ + Resource, error::OTelSdkError, metrics::{PeriodicReader, SdkMeterProvider}, }, @@ -164,7 +166,7 @@ pub(crate) fn client_name() -> &'static str { #[derive(Clone, Debug)] pub(crate) struct Observability { pub(crate) metrics: Option>, - common_attributes: [opentelemetry::KeyValue; 3], + common_attributes: [KeyValue; 3], meter_provider: Option>, } @@ -174,9 +176,9 @@ impl Observability { Self { metrics: None, common_attributes: [ - opentelemetry::KeyValue::new("client_uid", ""), - opentelemetry::KeyValue::new("client_name", ""), - opentelemetry::KeyValue::new("database", ""), + KeyValue::new("client_uid", ""), + KeyValue::new("client_name", ""), + KeyValue::new("database", ""), ], meter_provider: None, } @@ -240,13 +242,13 @@ impl Observability { let client_hash = generate_client_hash(&client_uid); let client_name = client_name(); - let resource = opentelemetry_sdk::Resource::builder() + let resource = Resource::builder() .with_attributes([ - opentelemetry::KeyValue::new("project_id", project_id.to_string()), - opentelemetry::KeyValue::new("instance_id", instance_id.to_string()), - opentelemetry::KeyValue::new("location", "global"), - opentelemetry::KeyValue::new("instance_config", "unknown"), - opentelemetry::KeyValue::new("client_hash", client_hash), + KeyValue::new("project_id", project_id.to_string()), + KeyValue::new("instance_id", instance_id.to_string()), + KeyValue::new("location", "global"), + KeyValue::new("instance_config", "unknown"), + KeyValue::new("client_hash", client_hash), ]) .build(); @@ -264,9 +266,9 @@ impl Observability { let metrics = SpannerMetrics::new(meter); let common_attributes = [ - opentelemetry::KeyValue::new("client_uid", client_uid), - opentelemetry::KeyValue::new("client_name", client_name), - opentelemetry::KeyValue::new("database", database_id.to_string()), + KeyValue::new("client_uid", client_uid), + KeyValue::new("client_name", client_name), + KeyValue::new("database", database_id.to_string()), ]; Self { @@ -281,9 +283,9 @@ impl Observability { Self { metrics: Some(Arc::new(metrics)), common_attributes: [ - opentelemetry::KeyValue::new("client_uid", "test-uid"), - opentelemetry::KeyValue::new("client_name", "test-name"), - opentelemetry::KeyValue::new("database", "test-db"), + KeyValue::new("client_uid", "test-uid"), + KeyValue::new("client_name", "test-name"), + KeyValue::new("database", "test-db"), ], meter_provider: Some(Arc::new(meter_provider)), } @@ -321,9 +323,9 @@ impl Observability { let status = result_to_status_str(result); let method_name = normalize_method_name(method); let attributes = [ - opentelemetry::KeyValue::new("method", method_name), - opentelemetry::KeyValue::new("status", status), - opentelemetry::KeyValue::new("directpath_enabled", "false"), + KeyValue::new("method", method_name), + KeyValue::new("status", status), + KeyValue::new("directpath_enabled", "false"), self.common_attributes[0].clone(), self.common_attributes[1].clone(), self.common_attributes[2].clone(), @@ -352,10 +354,10 @@ impl Observability { let status = error.map_or("OK", error_to_status_str); let method_name = normalize_method_name(method); let attributes = [ - opentelemetry::KeyValue::new("method", method_name), - opentelemetry::KeyValue::new("status", status), - opentelemetry::KeyValue::new("directpath_enabled", "false"), - opentelemetry::KeyValue::new("directpath_used", "false"), + KeyValue::new("method", method_name), + KeyValue::new("status", status), + KeyValue::new("directpath_enabled", "false"), + KeyValue::new("directpath_used", "false"), self.common_attributes[0].clone(), self.common_attributes[1].clone(), self.common_attributes[2].clone(), @@ -813,9 +815,9 @@ mod tests { let o11y = Observability { metrics: Some(Arc::new(metrics)), common_attributes: [ - opentelemetry::KeyValue::new("client_uid", ""), - opentelemetry::KeyValue::new("client_name", ""), - opentelemetry::KeyValue::new("database", ""), + KeyValue::new("client_uid", ""), + KeyValue::new("client_name", ""), + KeyValue::new("database", ""), ], meter_provider: Some(Arc::new(provider.clone())), }; @@ -1052,9 +1054,9 @@ mod tests { let o11y = Observability { metrics: Some(Arc::new(metrics)), common_attributes: [ - opentelemetry::KeyValue::new("client_uid", "test-uid"), - opentelemetry::KeyValue::new("client_name", "test-name"), - opentelemetry::KeyValue::new("database", "test-db"), + KeyValue::new("client_uid", "test-uid"), + KeyValue::new("client_name", "test-name"), + KeyValue::new("database", "test-db"), ], meter_provider: Some(Arc::new(provider.clone())), }; @@ -1233,8 +1235,8 @@ mod tests { let metrics = SpannerMetrics::new(meter); let attributes = [ - opentelemetry::KeyValue::new("method", "ExecuteSql"), - opentelemetry::KeyValue::new("status", "OK"), + KeyValue::new("method", "ExecuteSql"), + KeyValue::new("status", "OK"), ]; metrics.operation_latencies.record(12.5, &attributes); @@ -1330,9 +1332,9 @@ mod tests { let o11y = Observability { metrics: Some(Arc::new(metrics)), common_attributes: [ - opentelemetry::KeyValue::new("client_uid", "test-uid"), - opentelemetry::KeyValue::new("client_name", "spanner-rust/1.0.0"), - opentelemetry::KeyValue::new("database", "test-db"), + KeyValue::new("client_uid", "test-uid"), + KeyValue::new("client_name", "spanner-rust/1.0.0"), + KeyValue::new("database", "test-db"), ], meter_provider: Some(Arc::new(provider.clone())), }; @@ -1486,7 +1488,7 @@ mod tests { let o11y = Arc::new(Observability::for_test(metrics, provider.clone())); let interceptor = SpannerMetricsInterceptor; - let options = crate::RequestOptions::default().insert_extension(o11y.clone()); + let options = crate::RequestOptions::default().insert_extension(Arc::clone(&o11y)); let start_time = Instant::now(); // Attempt 1 fails with UNAVAILABLE diff --git a/src/spanner/src/request_id.rs b/src/spanner/src/request_id.rs index 840b4d760b..5524fdee90 100644 --- a/src/spanner/src/request_id.rs +++ b/src/spanner/src/request_id.rs @@ -86,12 +86,13 @@ impl RequestIdCreator { mod tests { use super::*; use crate::client::Spanner; + use crate::observability::Observability; use gaxi::grpc::tonic::{Response, Status}; use google_cloud_auth::credentials::anonymous::Builder as Anonymous; use google_cloud_test_macros::tokio_test_no_panics; use spanner_grpc_mock::google::spanner::v1 as mock_v1; use spanner_grpc_mock::{MockSpanner, start}; - use std::sync::Mutex; + use std::sync::{Arc, Mutex}; #[test] fn traits() { @@ -151,12 +152,12 @@ mod tests { #[tokio_test_no_panics] async fn request_id_header_sent_unary_rpc() { - let captured = std::sync::Arc::new(Mutex::new(Vec::new())); + let captured = Arc::new(Mutex::new(Vec::new())); let mut mock = MockSpanner::new(); let mut seq = mockall::Sequence::new(); // 1. Initial attempt of first RPC -> records header and returns UNAVAILABLE to force retry - let captured_clone = captured.clone(); + let captured_clone = Arc::clone(&captured); mock.expect_create_session() .once() .in_sequence(&mut seq) @@ -168,12 +169,15 @@ mod tests { .to_str() .expect("should be valid ASCII") .to_string(); - captured_clone.lock().unwrap().push(request_id); + captured_clone + .lock() + .expect("mutex lock should succeed") + .push(request_id); Err(Status::unavailable("server is unavailable")) }); // 2. Retry attempt of first RPC -> records header and returns Ok(Session) - let captured_clone = captured.clone(); + let captured_clone = Arc::clone(&captured); mock.expect_create_session() .once() .in_sequence(&mut seq) @@ -185,7 +189,10 @@ mod tests { .to_str() .expect("should be valid ASCII") .to_string(); - captured_clone.lock().unwrap().push(request_id); + captured_clone + .lock() + .expect("mutex lock should succeed") + .push(request_id); Ok(Response::new(mock_v1::Session { name: "projects/p/instances/i/databases/d/sessions/s1".to_string(), ..Default::default() @@ -193,7 +200,7 @@ mod tests { }); // 3. Second RPC -> records header and returns Ok(Session) - let captured_clone = captured.clone(); + let captured_clone = Arc::clone(&captured); mock.expect_create_session() .once() .in_sequence(&mut seq) @@ -205,7 +212,10 @@ mod tests { .to_str() .expect("should be valid ASCII") .to_string(); - captured_clone.lock().unwrap().push(request_id); + captured_clone + .lock() + .expect("mutex lock should succeed") + .push(request_id); Ok(Response::new(mock_v1::Session { name: "projects/p/instances/i/databases/d/sessions/s2".to_string(), ..Default::default() @@ -233,7 +243,7 @@ mod tests { request.clone(), crate::RequestOptions::default(), 0, - &crate::observability::Observability::disabled_arc(), + &Observability::disabled_arc(), ) .await .expect("first create_session should succeed after retry"); @@ -244,12 +254,12 @@ mod tests { request, crate::RequestOptions::default(), 0, - &crate::observability::Observability::disabled_arc(), + &Observability::disabled_arc(), ) .await .expect("second create_session should succeed"); - let ids = captured.lock().unwrap(); + let ids = captured.lock().expect("mutex lock should succeed"); assert_eq!(ids.len(), 3, "should have captured 3 RPC attempt headers"); let id_rpc1_attempt1 = &ids[0]; @@ -261,8 +271,14 @@ mod tests { "Request ID should start with version 1, got {id_rpc1_attempt1}" ); - let prefix1_attempt1 = id_rpc1_attempt1.rsplit_once('.').unwrap().0; - let prefix1_attempt2 = id_rpc1_attempt2.rsplit_once('.').unwrap().0; + let prefix1_attempt1 = id_rpc1_attempt1 + .rsplit_once('.') + .expect("should have dot separator") + .0; + let prefix1_attempt2 = id_rpc1_attempt2 + .rsplit_once('.') + .expect("should have dot separator") + .0; assert_eq!( prefix1_attempt2, prefix1_attempt1, "Retry attempt should have exactly the same values as initial attempt" @@ -303,12 +319,12 @@ mod tests { async fn request_id_header_sent_streaming_rpc() { use crate::result_set::tests::adapt; - let captured = std::sync::Arc::new(Mutex::new(Vec::new())); + let captured = Arc::new(Mutex::new(Vec::new())); let mut mock = MockSpanner::new(); let mut seq = mockall::Sequence::new(); // 1. Initial attempt of first streaming SQL -> records header and returns stream with 1 row + UNAVAILABLE - let captured_clone = captured.clone(); + let captured_clone = Arc::clone(&captured); mock.expect_execute_streaming_sql() .once() .in_sequence(&mut seq) @@ -320,7 +336,10 @@ mod tests { .to_str() .expect("should be valid ASCII") .to_string(); - captured_clone.lock().unwrap().push(request_id); + captured_clone + .lock() + .expect("mutex lock should succeed") + .push(request_id); let prs1 = mock_v1::PartialResultSet { metadata: Some(mock_v1::ResultSetMetadata { @@ -343,7 +362,7 @@ mod tests { }); // 2. Retry attempt of first streaming SQL -> records header and returns stream with row2 (last) - let captured_clone = captured.clone(); + let captured_clone = Arc::clone(&captured); mock.expect_execute_streaming_sql() .once() .in_sequence(&mut seq) @@ -355,7 +374,10 @@ mod tests { .to_str() .expect("should be valid ASCII") .to_string(); - captured_clone.lock().unwrap().push(request_id); + captured_clone + .lock() + .expect("mutex lock should succeed") + .push(request_id); let prs2 = mock_v1::PartialResultSet { values: vec![prost_types::Value { @@ -370,7 +392,7 @@ mod tests { }); // 3. Second streaming query -> records header and returns stream with row3 (last) - let captured_clone = captured.clone(); + let captured_clone = Arc::clone(&captured); mock.expect_execute_streaming_sql() .once() .in_sequence(&mut seq) @@ -382,7 +404,10 @@ mod tests { .to_str() .expect("should be valid ASCII") .to_string(); - captured_clone.lock().unwrap().push(request_id); + captured_clone + .lock() + .expect("mutex lock should succeed") + .push(request_id); let prs3 = mock_v1::PartialResultSet { metadata: Some(mock_v1::ResultSetMetadata { @@ -427,7 +452,7 @@ mod tests { .database_client("projects/p/instances/i/databases/d") .build() .await - .unwrap(); + .expect("database client build should succeed"); // Execute first streaming query via single_use().execute_query (attempt 1 -> row1 + UNAVAILABLE, attempt 2 -> row2) let mut rs1 = db_client @@ -451,7 +476,7 @@ mod tests { row.expect("row should succeed"); } - let ids = captured.lock().unwrap(); + let ids = captured.lock().expect("mutex lock should succeed"); assert_eq!(ids.len(), 3, "should have captured 3 RPC attempt headers"); let id_rpc1_attempt1 = &ids[0]; @@ -463,8 +488,14 @@ mod tests { "Request ID should start with version 1, got {id_rpc1_attempt1}" ); - let prefix1_attempt1 = id_rpc1_attempt1.rsplit_once('.').unwrap().0; - let prefix1_attempt2 = id_rpc1_attempt2.rsplit_once('.').unwrap().0; + let prefix1_attempt1 = id_rpc1_attempt1 + .rsplit_once('.') + .expect("should have dot separator") + .0; + let prefix1_attempt2 = id_rpc1_attempt2 + .rsplit_once('.') + .expect("should have dot separator") + .0; assert_eq!( prefix1_attempt2, prefix1_attempt1, "Retry attempt should have exactly the same values as initial attempt" @@ -505,7 +536,7 @@ mod tests { async fn request_id_header_sent_read_write_transaction_aborted_retry() { use crate::transaction_retry_policy::tests::create_aborted_status; - let captured = std::sync::Arc::new(Mutex::new(Vec::new())); + let captured = Arc::new(Mutex::new(Vec::new())); let mut mock = MockSpanner::new(); let mut seq = mockall::Sequence::new(); @@ -521,7 +552,7 @@ mod tests { }); // 1. First transaction attempt -> execute_sql fails with ABORTED - let captured_clone = captured.clone(); + let captured_clone = Arc::clone(&captured); mock.expect_execute_sql() .once() .in_sequence(&mut seq) @@ -533,12 +564,15 @@ mod tests { .to_str() .expect("should be valid ASCII") .to_string(); - captured_clone.lock().unwrap().push(request_id); + captured_clone + .lock() + .expect("mutex lock should succeed") + .push(request_id); Err(create_aborted_status(std::time::Duration::from_nanos(1))) }); // 2. Second transaction attempt (after ABORTED retry) -> execute_sql succeeds - let captured_clone = captured.clone(); + let captured_clone = Arc::clone(&captured); mock.expect_execute_sql() .once() .in_sequence(&mut seq) @@ -550,7 +584,10 @@ mod tests { .to_str() .expect("should be valid ASCII") .to_string(); - captured_clone.lock().unwrap().push(request_id); + captured_clone + .lock() + .expect("mutex lock should succeed") + .push(request_id); Ok(Response::new(mock_v1::ResultSet { metadata: Some(mock_v1::ResultSetMetadata { transaction: Some(mock_v1::Transaction { @@ -568,7 +605,7 @@ mod tests { }); // 3. Second transaction attempt -> commit succeeds - let captured_clone = captured.clone(); + let captured_clone = Arc::clone(&captured); mock.expect_commit() .once() .in_sequence(&mut seq) @@ -580,7 +617,10 @@ mod tests { .to_str() .expect("should be valid ASCII") .to_string(); - captured_clone.lock().unwrap().push(request_id); + captured_clone + .lock() + .expect("mutex lock should succeed") + .push(request_id); Ok(Response::new(mock_v1::CommitResponse::default())) }); @@ -598,10 +638,10 @@ mod tests { .database_client("projects/p/instances/i/databases/d") .build() .await - .unwrap(); + .expect("database client build should succeed"); - let count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); - let count_clone = count.clone(); + let count = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let count_clone = Arc::clone(&count); let runner = db_client .read_write_transaction() @@ -624,7 +664,7 @@ mod tests { "transaction closure should have run twice" ); - let ids = captured.lock().unwrap(); + let ids = captured.lock().expect("mutex lock should succeed"); assert_eq!( ids.len(), 3,