diff --git a/src/spanner/src/client.rs b/src/spanner/src/client.rs index 6e90743797..abe406035a 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(Arc::clone(o11y)); 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(), + &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(), + &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(), + &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(), + &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(), + &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(), + &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(), + &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(), + &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(), + &Observability::disabled_arc(), ) .await; diff --git a/src/spanner/src/observability/exporter.rs b/src/spanner/src/observability/exporter.rs index e5fbcb4ebc..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,25 +397,34 @@ 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 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"), - 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 = 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,20 +433,20 @@ 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"), - 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(); - 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 +478,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 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 = 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 +521,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() @@ -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"); @@ -542,7 +547,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 +597,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 +617,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,30 +641,23 @@ mod tests { } #[test] - fn test_is_permission_denied() { - let status_pd = google_cloud_gax::error::rpc::Status::default() - .set_code(google_cloud_gax::error::rpc::Code::PermissionDenied); + fn is_permission_denied() { + let status_pd = + google_cloud_gax::error::rpc::Status::default().set_code(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 status_nf = google_cloud_gax::error::rpc::Status::default().set_code(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() { - 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"); + fn value_to_string_all_variants() { + 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 4856f0d0aa..9a516c5808 100644 --- a/src/spanner/src/observability/metrics.rs +++ b/src/spanner/src/observability/metrics.rs @@ -14,22 +14,30 @@ 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::KeyValue, opentelemetry::metrics::{Counter, Histogram, Meter, MeterProvider}, opentelemetry_sdk::{ + Resource, error::OTelSdkError, metrics::{PeriodicReader, SdkMeterProvider}, }, + std::borrow::Cow, + std::sync::LazyLock, + std::time::Instant, }; #[cfg(feature = "_experimental-builtin-metrics")] @@ -44,7 +52,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 +60,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 +95,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 +108,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 +131,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 +143,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 +158,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")) } @@ -154,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>, } @@ -164,14 +176,19 @@ 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, } } + #[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 +198,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 +221,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(); } @@ -217,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(); @@ -241,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 { @@ -253,13 +278,27 @@ impl Observability { } } + #[cfg(test)] + pub(crate) fn for_test(metrics: SpannerMetrics, meter_provider: SdkMeterProvider) -> Self { + Self { + metrics: Some(Arc::new(metrics)), + common_attributes: [ + KeyValue::new("client_uid", "test-uid"), + KeyValue::new("client_name", "test-name"), + 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 +310,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("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"), 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("status", status), - opentelemetry::KeyValue::new("directpath_enabled", "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(), ]; 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 +399,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 +528,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 +542,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 +577,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 +598,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 +636,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 +675,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 +795,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] @@ -548,21 +815,25 @@ 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())), }; 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 +1006,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 +1019,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: [ + KeyValue::new("client_uid", "test-uid"), + KeyValue::new("client_name", "test-name"), + 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 +1201,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] @@ -810,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); @@ -834,8 +1259,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()); } } } @@ -907,21 +1332,20 @@ 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())), }; 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 +1366,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 +1423,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 +1457,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(Arc::clone(&o11y)); + 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..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(), + &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(), + &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, 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)