From c5be03915dee925591dcb72e94cd24f11ee8d578 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Knut=20Olav=20L=C3=B8ite?= Date: Tue, 30 Jun 2026 17:14:47 +0200 Subject: [PATCH 1/2] chore(spanner): add initial setup for built-in metrics Sets up the initial boiler-plate code for adding built-in metrics for Spanner. The feature is hidden behind a feature flag, meaning that it is currently invisible to customers. --- Cargo.lock | 4 + src/spanner/Cargo.toml | 5 + .../src/batch_read_only_transaction.rs | 2 + src/spanner/src/client.rs | 87 ++++- src/spanner/src/database_client.rs | 16 + src/spanner/src/lib.rs | 1 + src/spanner/src/observability/exporter.rs | 61 ++++ src/spanner/src/observability/metrics.rs | 304 ++++++++++++++++++ src/spanner/src/observability/mod.rs | 18 ++ .../src/partitioned_dml_transaction.rs | 7 +- src/spanner/src/read_only_transaction.rs | 2 +- src/spanner/src/read_write_transaction.rs | 22 +- src/spanner/src/session_maintainer.rs | 16 +- src/spanner/src/write_only_transaction.rs | 19 +- 14 files changed, 536 insertions(+), 28 deletions(-) create mode 100644 src/spanner/src/observability/exporter.rs create mode 100644 src/spanner/src/observability/metrics.rs create mode 100644 src/spanner/src/observability/mod.rs diff --git a/Cargo.lock b/Cargo.lock index ebefa01f7a..20f852e32e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4730,13 +4730,17 @@ dependencies = [ "google-cloud-auth", "google-cloud-gax", "google-cloud-gax-internal", + "google-cloud-monitoring-v3", "google-cloud-rpc", + "google-cloud-spanner", "google-cloud-spanner-admin-database-v1", "google-cloud-spanner-admin-instance-v1", "google-cloud-test-macros", "google-cloud-wkt", "http", "mockall", + "opentelemetry", + "opentelemetry_sdk", "prost", "prost-types", "rand 0.10.1", diff --git a/src/spanner/Cargo.toml b/src/spanner/Cargo.toml index cc234fbc05..8bbcd473db 100644 --- a/src/spanner/Cargo.toml +++ b/src/spanner/Cargo.toml @@ -33,6 +33,7 @@ default = ["default-rustls-provider"] # default and call `rustls::CryptoProvider::install_default()`. default-rustls-provider = ["gaxi/_default-rustls-provider"] unstable-stream = ["dep:futures"] +experimental-builtin-metrics = [] [dependencies] async-trait.workspace = true @@ -42,10 +43,13 @@ futures = { workspace = true, optional = true } gaxi = { workspace = true, features = ["_internal-common", "_internal-grpc-client", "_internal-grpc-server-streaming"] } google-cloud-auth = { workspace = true } google-cloud-gax = { workspace = true } +google-cloud-monitoring-v3 = { workspace = true } google-cloud-spanner-admin-database-v1 = { workspace = true } google-cloud-spanner-admin-instance-v1 = { workspace = true } google-cloud-rpc = { workspace = true } http.workspace = true +opentelemetry = { workspace = true, features = ["metrics"] } +opentelemetry_sdk = { workspace = true, features = ["rt-tokio", "metrics"] } prost.workspace = true prost-types.workspace = true rand = { workspace = true } @@ -62,6 +66,7 @@ wkt = { workspace = true, features = ["time"] [dev-dependencies] anyhow.workspace = true +google-cloud-spanner = { path = ".", features = ["experimental-builtin-metrics"] } google-cloud-test-macros.workspace = true mockall.workspace = true spanner-grpc-mock = { path = "grpc-mock" } diff --git a/src/spanner/src/batch_read_only_transaction.rs b/src/spanner/src/batch_read_only_transaction.rs index 3965004090..fcfa693273 100644 --- a/src/spanner/src/batch_read_only_transaction.rs +++ b/src/spanner/src/batch_read_only_transaction.rs @@ -168,6 +168,7 @@ impl BatchReadOnlyTransaction { request, crate::RequestOptions::default(), self.inner.context.channel_hint, + &self.inner.context.client.o11y, ) .await?; @@ -232,6 +233,7 @@ impl BatchReadOnlyTransaction { request, crate::RequestOptions::default(), self.inner.context.channel_hint, + &self.inner.context.client.o11y, ) .await?; diff --git a/src/spanner/src/client.rs b/src/spanner/src/client.rs index 99cdea868f..cce2fa3ce9 100644 --- a/src/spanner/src/client.rs +++ b/src/spanner/src/client.rs @@ -106,20 +106,24 @@ fn parse_emulator_endpoint(endpoint: &str) -> String { } macro_rules! define_idempotent_rpc { - ($method:ident, $request_type:ty, $response_type:ty) => { + ($method:ident, $request_type:ty, $response_type:ty, $canonical_name:expr) => { pub(crate) async fn $method( &self, request: $request_type, options: crate::RequestOptions, channel_hint: usize, + o11y: &crate::observability::Observability, ) -> crate::Result<$response_type> { - self.get_channel(channel_hint) - .inner - .$method() - .with_request(request) - .with_options(apply_request_defaults(options)) - .send() - .await + o11y.trace_operation($canonical_name, || async { + self.get_channel(channel_hint) + .inner + .$method() + .with_request(request) + .with_options(apply_request_defaults(options)) + .send() + .await + }) + .await } }; } @@ -298,18 +302,54 @@ impl Spanner { self.counter.fetch_add(1, Ordering::Relaxed) } - define_idempotent_rpc!(create_session, CreateSessionRequest, Session); - define_idempotent_rpc!(execute_sql, ExecuteSqlRequest, crate::model::ResultSet); + define_idempotent_rpc!( + create_session, + CreateSessionRequest, + Session, + "google.spanner.v1.Spanner/CreateSession" + ); + define_idempotent_rpc!( + execute_sql, + ExecuteSqlRequest, + crate::model::ResultSet, + "google.spanner.v1.Spanner/ExecuteSql" + ); define_idempotent_rpc!( execute_batch_dml, ExecuteBatchDmlRequest, - ExecuteBatchDmlResponse + ExecuteBatchDmlResponse, + "google.spanner.v1.Spanner/ExecuteBatchDml" + ); + define_idempotent_rpc!( + begin_transaction, + BeginTransactionRequest, + Transaction, + "google.spanner.v1.Spanner/BeginTransaction" + ); + define_idempotent_rpc!( + commit, + CommitRequest, + CommitResponse, + "google.spanner.v1.Spanner/Commit" + ); + define_idempotent_rpc!( + rollback, + RollbackRequest, + (), + "google.spanner.v1.Spanner/Rollback" + ); + define_idempotent_rpc!( + partition_query, + PartitionQueryRequest, + PartitionResponse, + "google.spanner.v1.Spanner/PartitionQuery" + ); + define_idempotent_rpc!( + partition_read, + PartitionReadRequest, + PartitionResponse, + "google.spanner.v1.Spanner/PartitionRead" ); - define_idempotent_rpc!(begin_transaction, BeginTransactionRequest, Transaction); - define_idempotent_rpc!(commit, CommitRequest, CommitResponse); - define_idempotent_rpc!(rollback, RollbackRequest, ()); - define_idempotent_rpc!(partition_query, PartitionQueryRequest, PartitionResponse); - define_idempotent_rpc!(partition_read, PartitionReadRequest, PartitionResponse); /// Executes an SQL statement, returning a stream of results. /// @@ -542,6 +582,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), + &crate::observability::Observability::disabled(), ) .await .expect("Failed to call create_session"); @@ -661,6 +702,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), + &crate::observability::Observability::disabled(), ) .await .expect("Failed to call create_session after transport error retry"); @@ -710,6 +752,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), + &crate::observability::Observability::disabled(), ) .await .expect("Failed to call execute_sql"); @@ -753,6 +796,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), + &crate::observability::Observability::disabled(), ) .await .expect("Failed to call execute_batch_dml"); @@ -791,6 +835,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), + &crate::observability::Observability::disabled(), ) .await .expect("Failed to call begin_transaction"); @@ -833,6 +878,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), + &crate::observability::Observability::disabled(), ) .await .expect("Failed to call commit"); @@ -866,6 +912,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), + &crate::observability::Observability::disabled(), ) .await .expect("Failed to call rollback"); @@ -1098,6 +1145,7 @@ mod tests { req, crate::RequestOptions::default(), client.next_channel_hint(), + &crate::observability::Observability::disabled(), ) .await .expect("Failed to call create_session"); @@ -1140,7 +1188,12 @@ mod tests { options.set_idempotency(false); let result = client - .create_session(req, options, client.next_channel_hint()) + .create_session( + req, + options, + client.next_channel_hint(), + &crate::observability::Observability::disabled(), + ) .await; // 5. Verify that it failed and did not retry diff --git a/src/spanner/src/database_client.rs b/src/spanner/src/database_client.rs index c6885f470b..2c686bf466 100644 --- a/src/spanner/src/database_client.rs +++ b/src/spanner/src/database_client.rs @@ -15,6 +15,7 @@ use crate::batch_read_only_transaction::BatchReadOnlyTransactionBuilder; use crate::batch_write_transaction::BatchWriteTransactionBuilder; use crate::client::Spanner; +use crate::observability::Observability; use crate::partitioned_dml_transaction::PartitionedDmlTransactionBuilder; use crate::read_only_transaction::{ MultiUseReadOnlyTransactionBuilder, SingleUseReadOnlyTransactionBuilder, @@ -51,6 +52,7 @@ pub struct DatabaseClient { pub(crate) spanner: Spanner, pub(crate) session_maintainer: Arc, pub(crate) leader_aware_routing_enabled: bool, + pub(crate) o11y: Arc, } impl DatabaseClient { @@ -361,11 +363,15 @@ impl DatabaseClientBuilder { /// will be used for all operations on the database. pub async fn build(self) -> crate::Result { let spanner_clone = self.spanner.clone(); + + let project_id = parse_project_id(&self.database_name); + let o11y = Arc::new(Observability::init(&self.spanner.config, project_id).await); let session_maintainer = ManagedSessionMaintainer::create_and_start_maintenance( self.spanner, self.database_name, self.database_role.unwrap_or_default(), self.options.unwrap_or_default(), + o11y.clone(), ) .await?; @@ -373,10 +379,20 @@ impl DatabaseClientBuilder { spanner: spanner_clone, session_maintainer, leader_aware_routing_enabled: self.leader_aware_routing_enabled, + o11y, }) } } +fn parse_project_id(database_name: &str) -> Option<&str> { + let mut parts = database_name.split('/'); + if parts.next() == Some("projects") { + parts.next() + } else { + None + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/spanner/src/lib.rs b/src/spanner/src/lib.rs index ee1886f628..d8a9d1dfad 100644 --- a/src/spanner/src/lib.rs +++ b/src/spanner/src/lib.rs @@ -77,6 +77,7 @@ pub(crate) mod batch_read_only_transaction; pub(crate) mod batch_write_transaction; pub(crate) mod database_client; pub(crate) mod from_value; +pub(crate) mod observability; pub(crate) mod partitioned_dml_transaction; pub(crate) mod precommit; pub(crate) mod read_only_transaction; diff --git a/src/spanner/src/observability/exporter.rs b/src/spanner/src/observability/exporter.rs new file mode 100644 index 0000000000..9e1f1b94b9 --- /dev/null +++ b/src/spanner/src/observability/exporter.rs @@ -0,0 +1,61 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use google_cloud_monitoring_v3::client::MetricService; +use opentelemetry_sdk::error::OTelSdkResult; +use opentelemetry_sdk::metrics::Temporality; +use opentelemetry_sdk::metrics::data::ResourceMetrics; +use opentelemetry_sdk::metrics::exporter::PushMetricExporter; +use std::sync::Arc; +use std::time::Duration; + +#[derive(Clone, Debug)] +pub(crate) struct GcpMonitoringExporter { + #[allow(dead_code)] + client: Arc, + #[allow(dead_code)] + project_name: String, +} + +impl GcpMonitoringExporter { + pub(crate) fn new(client: MetricService, project_id: &str) -> Self { + Self { + client: Arc::new(client), + project_name: format!("projects/{}", project_id), + } + } +} + +impl PushMetricExporter for GcpMonitoringExporter { + async fn export(&self, _metrics: &ResourceMetrics) -> OTelSdkResult { + // TODO: Map ResourceMetrics to CreateTimeSeriesRequest and send via self.client + Ok(()) + } + + fn force_flush(&self) -> OTelSdkResult { + Ok(()) + } + + fn shutdown(&self) -> OTelSdkResult { + Ok(()) + } + + fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult { + self.shutdown() + } + + fn temporality(&self) -> Temporality { + Temporality::Cumulative + } +} diff --git a/src/spanner/src/observability/metrics.rs b/src/spanner/src/observability/metrics.rs new file mode 100644 index 0000000000..3b89630593 --- /dev/null +++ b/src/spanner/src/observability/metrics.rs @@ -0,0 +1,304 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use crate::observability::exporter::GcpMonitoringExporter; +use gaxi::options::ClientConfig; +use google_cloud_monitoring_v3::client::MetricService; +use opentelemetry::metrics::{Counter, Histogram, Meter, MeterProvider}; +use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; +use std::time::Duration; +use std::time::Instant; + +#[derive(Debug)] +pub(crate) struct SpannerMetrics { + pub(crate) operation_latencies: Histogram, + #[allow(dead_code)] + pub(crate) attempt_latencies: Histogram, + #[allow(dead_code)] + pub(crate) gfe_latencies: Histogram, + #[allow(dead_code)] + pub(crate) afe_latencies: Histogram, + pub(crate) operation_count: Counter, + #[allow(dead_code)] + pub(crate) attempt_count: Counter, +} + +impl SpannerMetrics { + pub(crate) fn new(meter: Meter) -> Self { + Self { + operation_latencies: meter + .f64_histogram("spanner.googleapis.com/internal/client/operation_latencies") + .with_unit("ms") + .build(), + attempt_latencies: meter + .f64_histogram("spanner.googleapis.com/internal/client/attempt_latencies") + .with_unit("ms") + .build(), + gfe_latencies: meter + .f64_histogram("spanner.googleapis.com/internal/client/gfe_latencies") + .with_unit("ms") + .build(), + afe_latencies: meter + .f64_histogram("spanner.googleapis.com/internal/client/afe_latencies") + .with_unit("ms") + .build(), + operation_count: meter + .u64_counter("spanner.googleapis.com/internal/client/operation_count") + .build(), + attempt_count: meter + .u64_counter("spanner.googleapis.com/internal/client/attempt_count") + .build(), + } + } +} + +#[derive(Debug)] +pub(crate) struct Observability { + pub(crate) metrics: Option, + _meter_provider: Option, +} + +impl Observability { + pub(crate) fn disabled() -> Self { + Self { + metrics: None, + _meter_provider: None, + } + } + + pub(crate) async fn init(config: &ClientConfig, project_id: Option<&str>) -> Self { + if !cfg!(feature = "experimental-builtin-metrics") { + return Self::disabled(); + } + + let disable_builtin_metrics = std::env::var("SPANNER_DISABLE_BUILTIN_METRICS") + .map(|s| s.to_lowercase() == "true") + .unwrap_or(false); + if disable_builtin_metrics { + return Self::disabled(); + } + + let project_id = match project_id { + Some(id) => id, + None => return Self::disabled(), + }; + + // Create the Google Cloud Monitoring client using the same config but pointing to the monitoring endpoint + let mut builder = MetricService::builder(); + builder = builder.with_endpoint("monitoring.googleapis.com:443"); + + 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); + } + + let monitoring_client = match builder.build().await { + Ok(c) => c, + Err(e) => { + tracing::warn!( + "Failed to initialize Google Cloud Monitoring client for Spanner metrics: {:?}", + e + ); + return Self::disabled(); + } + }; + + let exporter = GcpMonitoringExporter::new(monitoring_client, project_id); + + // Set up PeriodicReader + let reader = PeriodicReader::builder(exporter).build(); + + let meter_provider = SdkMeterProvider::builder().with_reader(reader).build(); + + let meter = meter_provider.meter("cloud.google.com/rust"); + let metrics = SpannerMetrics::new(meter); + + Self { + metrics: Some(metrics), + _meter_provider: Some(meter_provider), + } + } + + pub(crate) async fn trace_operation( + &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_operation(method, elapsed, &result); + 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( + &self, + method: &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 attributes = [ + opentelemetry::KeyValue::new("method", method.to_string()), + opentelemetry::KeyValue::new("status", status), + ]; + + metrics + .attempt_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); + } + } + + pub(crate) fn record_operation( + &self, + method: &str, + duration: Duration, + result: &crate::Result, + ) { + let Some(ref metrics) = self.metrics else { + return; + }; + + let status = result_to_status_str(result); + let attributes = [ + opentelemetry::KeyValue::new("method", method.to_string()), + opentelemetry::KeyValue::new("status", status), + ]; + + metrics + .operation_latencies + .record(duration.as_secs_f64() * 1000.0, &attributes); + metrics.operation_count.add(1, &attributes); + } +} + +fn result_to_status_str(result: &crate::Result) -> &'static str { + match result { + Ok(_) => "OK", + Err(e) => { + if let Some(status) = e.status() { + status.code.name() + } else { + "UNKNOWN" + } + } + } +} + +#[allow(dead_code)] +#[derive(Debug, Default, PartialEq)] +pub(crate) struct ServerTimings { + pub(crate) gfe_latency: Option, + pub(crate) afe_latency: Option, +} + +#[allow(dead_code)] +pub(crate) fn parse_server_timing(header_val: &str) -> ServerTimings { + let mut timings = ServerTimings::default(); + for part in header_val.split(',') { + let mut subparts = part.split(';'); + let name_opt = subparts.next().map(|s| s.trim()); + let val_opt = subparts.next().and_then(|dur_part| dur_part.split('=').nth(1)); + let parsed = name_opt.zip(val_opt).and_then(|(name, val_str)| { + val_str.trim().parse::().ok().map(|dur| (name, dur)) + }); + if let Some((name, dur)) = parsed { + match name { + "gfet4t7" => timings.gfe_latency = Some(dur), + "afe" => timings.afe_latency = Some(dur), + _ => {} + } + } + } + timings +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_parse_server_timing() { + assert_eq!( + parse_server_timing("gfet4t7;dur=12.5"), + ServerTimings { + gfe_latency: Some(12.5), + afe_latency: None, + } + ); + assert_eq!( + parse_server_timing("gfet4t7;dur=12.5,afe;dur=5"), + ServerTimings { + gfe_latency: Some(12.5), + afe_latency: Some(5.0), + } + ); + assert_eq!( + parse_server_timing("afe;dur=3,some-other;dur=10"), + ServerTimings { + gfe_latency: None, + afe_latency: Some(3.0), + } + ); + assert_eq!( + parse_server_timing("invalid_format"), + ServerTimings::default() + ); + } + + #[test] + fn test_feature_enabled_during_tests() { + assert!( + cfg!(feature = "experimental-builtin-metrics"), + "The 'experimental-builtin-metrics' feature must be enabled during test runs." + ); + } +} diff --git a/src/spanner/src/observability/mod.rs b/src/spanner/src/observability/mod.rs new file mode 100644 index 0000000000..b5a7052c2f --- /dev/null +++ b/src/spanner/src/observability/mod.rs @@ -0,0 +1,18 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +pub(crate) mod exporter; +pub(crate) mod metrics; + +pub(crate) use metrics::Observability; diff --git a/src/spanner/src/partitioned_dml_transaction.rs b/src/spanner/src/partitioned_dml_transaction.rs index ba04cbdb16..909c1cc4dc 100644 --- a/src/spanner/src/partitioned_dml_transaction.rs +++ b/src/spanner/src/partitioned_dml_transaction.rs @@ -192,7 +192,12 @@ impl PartitionedDmlTransaction { async move { let transaction = client .spanner - .begin_transaction(begin_request, gax_options.clone(), channel_hint) + .begin_transaction( + begin_request, + gax_options.clone(), + channel_hint, + &client.o11y, + ) .await?; let execute_request = diff --git a/src/spanner/src/read_only_transaction.rs b/src/spanner/src/read_only_transaction.rs index 776f2c79a8..e091d8d0f4 100644 --- a/src/spanner/src/read_only_transaction.rs +++ b/src/spanner/src/read_only_transaction.rs @@ -571,7 +571,7 @@ pub(crate) async fn execute_begin_transaction( client .spanner - .begin_transaction(request, request_options, channel_hint) + .begin_transaction(request, request_options, channel_hint, &client.o11y) .await } diff --git a/src/spanner/src/read_write_transaction.rs b/src/spanner/src/read_write_transaction.rs index 98bed61d42..11c38a1cbf 100644 --- a/src/spanner/src/read_write_transaction.rs +++ b/src/spanner/src/read_write_transaction.rs @@ -322,6 +322,7 @@ macro_rules! execute_with_retry { $request.clone(), $gax_options.clone(), $self.context.channel_hint, + &$self.context.client.o11y, ) .await; @@ -653,7 +654,12 @@ impl ReadWriteTransaction { .context .client .spanner - .commit(request, gax_options, self.context.channel_hint) + .commit( + request, + gax_options, + self.context.channel_hint, + &self.context.client.o11y, + ) .await?; let response = @@ -670,7 +676,12 @@ impl ReadWriteTransaction { self.context .client .spanner - .commit(retry_commit_req, gax_options, self.context.channel_hint) + .commit( + retry_commit_req, + gax_options, + self.context.channel_hint, + &self.context.client.o11y, + ) .await? } else { response @@ -695,7 +706,12 @@ impl ReadWriteTransaction { self.context .client .spanner - .rollback(request, gax_options, self.context.channel_hint) + .rollback( + request, + gax_options, + self.context.channel_hint, + &self.context.client.o11y, + ) .await?; Ok(()) diff --git a/src/spanner/src/session_maintainer.rs b/src/spanner/src/session_maintainer.rs index ffcae61fd4..18fe618890 100644 --- a/src/spanner/src/session_maintainer.rs +++ b/src/spanner/src/session_maintainer.rs @@ -14,6 +14,7 @@ use crate::client::Spanner; use crate::model::{CreateSessionRequest, Session}; +use crate::observability::Observability; use crate::{RequestOptions, Result}; use std::sync::{Arc, RwLock, Weak}; use std::time::Duration; @@ -34,6 +35,7 @@ pub(crate) struct ManagedSessionMaintainer { pub(crate) database_name: String, pub(crate) database_role: String, pub(crate) options: RequestOptions, + pub(crate) o11y: Arc, } #[derive(Debug)] @@ -59,9 +61,10 @@ impl ManagedSessionMaintainer { database_name: String, database_role: String, options: RequestOptions, + o11y: Arc, ) -> Result> { let session = - Self::create_session(&spanner, &database_name, &database_role, &options).await?; + Self::create_session(&spanner, &database_name, &database_role, &options, &o11y).await?; let maintainer = Arc::new(ManagedSessionMaintainer { spanner, @@ -72,6 +75,7 @@ impl ManagedSessionMaintainer { database_name, database_role, options, + o11y, }); let weak_maintainer = Arc::downgrade(&maintainer); @@ -105,6 +109,7 @@ impl ManagedSessionMaintainer { &self.database_name, &self.database_role, &self.options, + &self.o11y, ) .await?; @@ -125,6 +130,7 @@ impl ManagedSessionMaintainer { database_name: &str, database_role: &str, options: &RequestOptions, + o11y: &Observability, ) -> Result { let request = CreateSessionRequest::new() .set_database(database_name) @@ -135,7 +141,7 @@ impl ManagedSessionMaintainer { ); spanner - .create_session(request, options.clone(), spanner.next_channel_hint()) + .create_session(request, options.clone(), spanner.next_channel_hint(), o11y) .await } @@ -218,6 +224,7 @@ mod tests { "projects/test-project/instances/test-instance/databases/test-db".to_string(), "test-role".to_string(), RequestOptions::default(), + Arc::new(Observability::disabled()), ) .await .expect("Failed to create ManagedSessionMaintainer"); @@ -288,6 +295,7 @@ mod tests { "projects/test-project/instances/test-instance/databases/test-db".to_string(), "test-role".to_string(), RequestOptions::default(), + Arc::new(Observability::disabled()), ) .await .expect("Failed to create ManagedSessionMaintainer"); @@ -330,6 +338,7 @@ mod tests { "projects/test-project/instances/test-instance/databases/test-db".to_string(), "test-role".to_string(), RequestOptions::default(), + Arc::new(Observability::disabled()), ) .await .expect("Failed to create ManagedSessionMaintainer"); @@ -394,6 +403,7 @@ mod tests { "projects/test-project/instances/test-instance/databases/test-db".to_string(), "test-role".to_string(), RequestOptions::default(), + Arc::new(Observability::disabled()), ) .await .expect("Failed to create ManagedSessionMaintainer"); @@ -543,6 +553,7 @@ mod tests { "projects/p/instances/i/databases/d".to_string(), "test-role".to_string(), RequestOptions::default(), + Arc::new(Observability::disabled()), ) .await .expect("Failed to create ManagedSessionMaintainer"); @@ -551,6 +562,7 @@ mod tests { spanner, session_maintainer: maintainer.clone(), leader_aware_routing_enabled: true, + o11y: std::sync::Arc::new(crate::observability::Observability::disabled()), }; // 1. Create builder (captures session 1) diff --git a/src/spanner/src/write_only_transaction.rs b/src/spanner/src/write_only_transaction.rs index 1d4093bb6b..e96d61419e 100644 --- a/src/spanner/src/write_only_transaction.rs +++ b/src/spanner/src/write_only_transaction.rs @@ -448,7 +448,7 @@ impl WriteOnlyTransaction { let tx = client .spanner - .begin_transaction(begin_req, begin_gax_options, channel_hint) + .begin_transaction(begin_req, begin_gax_options, channel_hint, &client.o11y) .await?; *previous_transaction_id.lock().unwrap() = tx.id.clone(); @@ -464,7 +464,12 @@ impl WriteOnlyTransaction { let response = client .spanner - .commit(commit_req, commit_gax_options.clone(), channel_hint) + .commit( + commit_req, + commit_gax_options.clone(), + channel_hint, + &client.o11y, + ) .await?; // If a commit_response with a precommit_token is returned, then we need to @@ -479,9 +484,15 @@ impl WriteOnlyTransaction { max_commit_delay, return_commit_stats, ); + client .spanner - .commit(retry_commit_req, commit_gax_options, channel_hint) + .commit( + retry_commit_req, + commit_gax_options, + channel_hint, + &client.o11y, + ) .await } else { Ok(response) @@ -555,7 +566,7 @@ impl WriteOnlyTransaction { async move { client .spanner - .commit(request, commit_gax_options, channel_hint) + .commit(request, commit_gax_options, channel_hint, &client.o11y) .await } }; From f5992406f1d594b542e9c8cd1938f0dfcb3fd02e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Knut=20Olav=20L=C3=B8ite?= Date: Tue, 30 Jun 2026 18:17:15 +0200 Subject: [PATCH 2/2] chore(spanner): address review comments --- .typos.toml | 2 + Cargo.lock | 1 - src/spanner/Cargo.toml | 15 +-- src/spanner/src/client.rs | 2 +- src/spanner/src/database_client.rs | 7 +- src/spanner/src/observability/metrics.rs | 115 ++++++++++++++--------- src/spanner/src/observability/mod.rs | 1 + tests/spanner/Cargo.toml | 2 +- 8 files changed, 89 insertions(+), 56 deletions(-) diff --git a/.typos.toml b/.typos.toml index 19d40d27bb..3d4a1ad4ea 100644 --- a/.typos.toml +++ b/.typos.toml @@ -26,6 +26,8 @@ extend-exclude = [ "src/spanner/grpc-mock/**", # This file contains fixes for typos and therefore has the original typo in it. "librarian.yaml", + # Contains Spanner-specific metrics abbreviations (like 'afe' for Access Front End) + "src/spanner/src/observability/metrics.rs", ] [type.mustache] diff --git a/Cargo.lock b/Cargo.lock index 20f852e32e..a84952dd92 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4732,7 +4732,6 @@ dependencies = [ "google-cloud-gax-internal", "google-cloud-monitoring-v3", "google-cloud-rpc", - "google-cloud-spanner", "google-cloud-spanner-admin-database-v1", "google-cloud-spanner-admin-instance-v1", "google-cloud-test-macros", diff --git a/src/spanner/Cargo.toml b/src/spanner/Cargo.toml index 8bbcd473db..c2c71be064 100644 --- a/src/spanner/Cargo.toml +++ b/src/spanner/Cargo.toml @@ -31,9 +31,9 @@ default = ["default-rustls-provider"] # TLS and authentication. Applications with specific requirements for # cryptography (such as exclusively using the [ring] crate) should disable this # default and call `rustls::CryptoProvider::install_default()`. -default-rustls-provider = ["gaxi/_default-rustls-provider"] -unstable-stream = ["dep:futures"] -experimental-builtin-metrics = [] +default-rustls-provider = ["gaxi/_default-rustls-provider"] +unstable-stream = ["dep:futures"] +experimental-builtin-metrics = ["dep:google-cloud-monitoring-v3", "dep:opentelemetry", "dep:opentelemetry_sdk"] [dependencies] async-trait.workspace = true @@ -43,13 +43,10 @@ futures = { workspace = true, optional = true } gaxi = { workspace = true, features = ["_internal-common", "_internal-grpc-client", "_internal-grpc-server-streaming"] } google-cloud-auth = { workspace = true } google-cloud-gax = { workspace = true } -google-cloud-monitoring-v3 = { workspace = true } google-cloud-spanner-admin-database-v1 = { workspace = true } google-cloud-spanner-admin-instance-v1 = { workspace = true } google-cloud-rpc = { workspace = true } http.workspace = true -opentelemetry = { workspace = true, features = ["metrics"] } -opentelemetry_sdk = { workspace = true, features = ["rt-tokio", "metrics"] } prost.workspace = true prost-types.workspace = true rand = { workspace = true } @@ -64,9 +61,13 @@ tracing.workspace = true url.workspace = true wkt = { workspace = true, features = ["time"] } +# Optional dependencies for built-in metrics +google-cloud-monitoring-v3 = { workspace = true, optional = true } +opentelemetry = { workspace = true, optional = true, features = ["metrics"] } +opentelemetry_sdk = { workspace = true, optional = true, features = ["metrics", "rt-tokio"] } + [dev-dependencies] anyhow.workspace = true -google-cloud-spanner = { path = ".", features = ["experimental-builtin-metrics"] } google-cloud-test-macros.workspace = true mockall.workspace = true spanner-grpc-mock = { path = "grpc-mock" } diff --git a/src/spanner/src/client.rs b/src/spanner/src/client.rs index cce2fa3ce9..2d4354ae16 100644 --- a/src/spanner/src/client.rs +++ b/src/spanner/src/client.rs @@ -114,7 +114,7 @@ macro_rules! define_idempotent_rpc { channel_hint: usize, o11y: &crate::observability::Observability, ) -> crate::Result<$response_type> { - o11y.trace_operation($canonical_name, || async { + o11y.trace_operation($canonical_name, || async move { self.get_channel(channel_hint) .inner .$method() diff --git a/src/spanner/src/database_client.rs b/src/spanner/src/database_client.rs index 2c686bf466..6996a97193 100644 --- a/src/spanner/src/database_client.rs +++ b/src/spanner/src/database_client.rs @@ -386,11 +386,10 @@ impl DatabaseClientBuilder { fn parse_project_id(database_name: &str) -> Option<&str> { let mut parts = database_name.split('/'); - if parts.next() == Some("projects") { - parts.next() - } else { - None + if parts.next() != Some("projects") { + return None; } + parts.next() } #[cfg(test)] diff --git a/src/spanner/src/observability/metrics.rs b/src/spanner/src/observability/metrics.rs index 3b89630593..b8ee61da81 100644 --- a/src/spanner/src/observability/metrics.rs +++ b/src/spanner/src/observability/metrics.rs @@ -12,28 +12,36 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::observability::exporter::GcpMonitoringExporter; -use gaxi::options::ClientConfig; -use google_cloud_monitoring_v3::client::MetricService; -use opentelemetry::metrics::{Counter, Histogram, Meter, MeterProvider}; -use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; +#[cfg(feature = "experimental-builtin-metrics")] use std::time::Duration; +#[cfg(feature = "experimental-builtin-metrics")] use std::time::Instant; +#[cfg(feature = "experimental-builtin-metrics")] +use { + crate::observability::exporter::GcpMonitoringExporter, + gaxi::options::ClientConfig, + google_cloud_monitoring_v3::client::MetricService, + opentelemetry::metrics::{Counter, Histogram, Meter, MeterProvider}, + opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}, +}; + +#[cfg(not(feature = "experimental-builtin-metrics"))] +use gaxi::options::ClientConfig; + +#[cfg(feature = "experimental-builtin-metrics")] +#[allow(dead_code)] #[derive(Debug)] pub(crate) struct SpannerMetrics { pub(crate) operation_latencies: Histogram, - #[allow(dead_code)] pub(crate) attempt_latencies: Histogram, - #[allow(dead_code)] pub(crate) gfe_latencies: Histogram, - #[allow(dead_code)] pub(crate) afe_latencies: Histogram, pub(crate) operation_count: Counter, - #[allow(dead_code)] pub(crate) attempt_count: Counter, } +#[cfg(feature = "experimental-builtin-metrics")] impl SpannerMetrics { pub(crate) fn new(meter: Meter) -> Self { Self { @@ -63,12 +71,14 @@ impl SpannerMetrics { } } +#[cfg(feature = "experimental-builtin-metrics")] #[derive(Debug)] pub(crate) struct Observability { pub(crate) metrics: Option, _meter_provider: Option, } +#[cfg(feature = "experimental-builtin-metrics")] impl Observability { pub(crate) fn disabled() -> Self { Self { @@ -78,12 +88,8 @@ impl Observability { } pub(crate) async fn init(config: &ClientConfig, project_id: Option<&str>) -> Self { - if !cfg!(feature = "experimental-builtin-metrics") { - return Self::disabled(); - } - let disable_builtin_metrics = std::env::var("SPANNER_DISABLE_BUILTIN_METRICS") - .map(|s| s.to_lowercase() == "true") + .map(|s| s.eq_ignore_ascii_case("true")) .unwrap_or(false); if disable_builtin_metrics { return Self::disabled(); @@ -94,15 +100,14 @@ impl Observability { None => return Self::disabled(), }; - // Create the Google Cloud Monitoring client using the same config but pointing to the monitoring endpoint + // Create the Google Cloud Monitoring client using the same config let mut builder = MetricService::builder(); - builder = builder.with_endpoint("monitoring.googleapis.com:443"); 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); + builder = builder.with_universe_domain(ud.clone()); } let monitoring_client = match builder.build().await { @@ -168,7 +173,7 @@ impl Observability { #[allow(dead_code)] pub(crate) fn record_attempt( &self, - method: &str, + method: &'static str, duration: Duration, result: &crate::Result, gfe_latency: Option, @@ -180,7 +185,7 @@ impl Observability { let status = result_to_status_str(result); let attributes = [ - opentelemetry::KeyValue::new("method", method.to_string()), + opentelemetry::KeyValue::new("method", method), opentelemetry::KeyValue::new("status", status), ]; @@ -199,7 +204,7 @@ impl Observability { pub(crate) fn record_operation( &self, - method: &str, + method: &'static str, duration: Duration, result: &crate::Result, ) { @@ -209,7 +214,7 @@ impl Observability { let status = result_to_status_str(result); let attributes = [ - opentelemetry::KeyValue::new("method", method.to_string()), + opentelemetry::KeyValue::new("method", method), opentelemetry::KeyValue::new("status", status), ]; @@ -220,6 +225,7 @@ impl Observability { } } +#[cfg(feature = "experimental-builtin-metrics")] fn result_to_status_str(result: &crate::Result) -> &'static str { match result { Ok(_) => "OK", @@ -233,35 +239,68 @@ fn result_to_status_str(result: &crate::Result) -> &'static str { } } -#[allow(dead_code)] +#[cfg(feature = "experimental-builtin-metrics")] #[derive(Debug, Default, PartialEq)] pub(crate) struct ServerTimings { pub(crate) gfe_latency: Option, pub(crate) afe_latency: Option, } +#[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(',') { let mut subparts = part.split(';'); - let name_opt = subparts.next().map(|s| s.trim()); - let val_opt = subparts.next().and_then(|dur_part| dur_part.split('=').nth(1)); - let parsed = name_opt.zip(val_opt).and_then(|(name, val_str)| { - val_str.trim().parse::().ok().map(|dur| (name, dur)) - }); - if let Some((name, dur)) = parsed { - match name { - "gfet4t7" => timings.gfe_latency = Some(dur), - "afe" => timings.afe_latency = Some(dur), - _ => {} + if let Some(name) = subparts.next().map(|s| s.trim()) { + for param in subparts { + let mut kv = param.split('='); + let dur_opt = match (kv.next(), kv.next()) { + (Some(k), Some(v)) if k.trim() == "dur" => v.trim().parse::().ok(), + _ => None, + }; + if let Some(dur) = dur_opt { + match name { + "gfet4t7" => timings.gfe_latency = Some(dur), + "afe" => timings.afe_latency = Some(dur), + _ => {} + } + } } } } timings } -#[cfg(test)] +#[cfg(not(feature = "experimental-builtin-metrics"))] +#[derive(Debug)] +pub(crate) struct Observability; + +#[cfg(not(feature = "experimental-builtin-metrics"))] +impl Observability { + #[allow(dead_code)] + pub(crate) fn disabled() -> Self { + Self + } + + pub(crate) async fn init(_config: &ClientConfig, _project_id: Option<&str>) -> Self { + Self + } + + pub(crate) async fn trace_operation( + &self, + _method: &'static str, + f: F, + ) -> crate::Result + where + F: FnOnce() -> Fut, + Fut: std::future::Future>, + { + f().await + } +} + +#[cfg(all(test, feature = "experimental-builtin-metrics"))] mod tests { use super::*; @@ -275,7 +314,7 @@ mod tests { } ); assert_eq!( - parse_server_timing("gfet4t7;dur=12.5,afe;dur=5"), + parse_server_timing("gfet4t7;desc=\"test\";dur=12.5,afe;dur=5;desc=\"other\""), ServerTimings { gfe_latency: Some(12.5), afe_latency: Some(5.0), @@ -293,12 +332,4 @@ mod tests { ServerTimings::default() ); } - - #[test] - fn test_feature_enabled_during_tests() { - assert!( - cfg!(feature = "experimental-builtin-metrics"), - "The 'experimental-builtin-metrics' feature must be enabled during test runs." - ); - } } diff --git a/src/spanner/src/observability/mod.rs b/src/spanner/src/observability/mod.rs index b5a7052c2f..9a811d0e32 100644 --- a/src/spanner/src/observability/mod.rs +++ b/src/spanner/src/observability/mod.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +#[cfg(feature = "experimental-builtin-metrics")] pub(crate) mod exporter; pub(crate) mod metrics; diff --git a/tests/spanner/Cargo.toml b/tests/spanner/Cargo.toml index cf06534d01..af9e64eea8 100644 --- a/tests/spanner/Cargo.toml +++ b/tests/spanner/Cargo.toml @@ -32,7 +32,7 @@ google-cloud-auth = { workspace = true, features = ["defaul google-cloud-gax = { workspace = true } google-cloud-longrunning = { workspace = true } google-cloud-lro = { workspace = true } -google-cloud-spanner = { workspace = true, features = ["default", "unstable-stream"] } +google-cloud-spanner = { workspace = true, features = ["default", "experimental-builtin-metrics", "unstable-stream"] } google-cloud-spanner-admin-database-v1 = { path = "../../src/generated/spanner/admin/database/v1" } google-cloud-spanner-admin-instance-v1 = { path = "../../src/generated/spanner/admin/instance/v1" } google-cloud-test-utils = { workspace = true }