diff --git a/Cargo.lock b/Cargo.lock index 60550c8a35..a01ddf9971 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5282,6 +5282,7 @@ dependencies = [ "serde", "serde_json", "serde_with", + "serial_test", "spanner-grpc-mock", "static_assertions", "thiserror", diff --git a/src/spanner/Cargo.toml b/src/spanner/Cargo.toml index 609d5b3915..a5d0bf2703 100644 --- a/src/spanner/Cargo.toml +++ b/src/spanner/Cargo.toml @@ -39,6 +39,7 @@ _experimental-builtin-metrics = [ "dep:opentelemetry", "dep:opentelemetry_sdk", "dep:uuid", + "gaxi/_internal-http-client", ] [dependencies] @@ -81,6 +82,7 @@ google-cloud-test-macros.workspace = true mockall.workspace = true opentelemetry_sdk = { workspace = true, features = ["metrics", "testing"] } scoped-env.workspace = true +serial_test.workspace = true spanner-grpc-mock = { path = "grpc-mock" } static_assertions.workspace = true tokio = { workspace = true, features = ["test-util"] } diff --git a/src/spanner/src/observability/metrics.rs b/src/spanner/src/observability/metrics.rs index 21c98eba7e..f1bb14248c 100644 --- a/src/spanner/src/observability/metrics.rs +++ b/src/spanner/src/observability/metrics.rs @@ -25,6 +25,7 @@ use std::time::Duration; use { crate::observability::exporter::GcpMonitoringExporter, gaxi::attempt_interceptor::AttemptInterceptor, + gaxi::http::reqwest::{Client, Url}, google_cloud_gax::options::RequestOptions, google_cloud_monitoring_v3::client::MetricService, http::header::{HeaderName, HeaderValue}, @@ -36,8 +37,12 @@ use { metrics::{PeriodicReader, SdkMeterProvider}, }, std::borrow::Cow, + std::env, + std::process, std::sync::LazyLock, std::time::Instant, + tokio::sync::OnceCell, + uuid::Uuid, }; #[cfg(feature = "_experimental-builtin-metrics")] @@ -51,6 +56,19 @@ pub(crate) const BUCKET_BOUNDARIES: [f64; 50] = [ 100000.0, 200000.0, 400000.0, 800000.0, 1600000.0, 3200000.0, ]; +#[cfg(feature = "_experimental-builtin-metrics")] +const DEFAULT_CLIENT_LOCATION: &str = "global"; +#[cfg(feature = "_experimental-builtin-metrics")] +const DEFAULT_GCP_CHECK_TIMEOUT_MS: u64 = 5000; +#[cfg(feature = "_experimental-builtin-metrics")] +const DEFAULT_GCP_CHECK_CONNECT_TIMEOUT_MS: u64 = 250; +#[cfg(feature = "_experimental-builtin-metrics")] +const GCE_METADATA_HOST_ENV_VAR: &str = "GCE_METADATA_HOST"; +#[cfg(feature = "_experimental-builtin-metrics")] +const DEFAULT_METADATA_ROOT: &str = "http://metadata.google.internal"; +#[cfg(feature = "_experimental-builtin-metrics")] +const INSTANCE_ZONE_METADATA_PATH: &str = "/computeMetadata/v1/instance/zone"; + #[cfg(feature = "_experimental-builtin-metrics")] #[derive(Debug)] pub(crate) struct SpannerMetrics { @@ -132,16 +150,19 @@ pub(crate) fn parse_database_name(database_name: &str) -> Option<(&str, &str, &s /// `UUID@PID@hostname`. #[cfg(feature = "_experimental-builtin-metrics")] pub(crate) fn generate_client_uid() -> String { - let uuid = uuid::Uuid::new_v4().to_string(); - let pid = std::process::id(); - let hostname = std::env::var("HOSTNAME") - .or_else(|_| std::env::var("COMPUTERNAME")) + let uuid = Uuid::new_v4().to_string(); + let pid = process::id(); + let hostname = env::var("HOSTNAME") + .or_else(|_| env::var("COMPUTERNAME")) .unwrap_or_else(|_| "localhost".to_string()); format!("{uuid}@{pid}@{hostname}") } /// 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`. +/// resource label using the 10 most significant bits of an FNV-1a 64-bit hash of `client_uid`. +/// +/// The 10-bit prefix (values in range `[000000, 0003ff]`) intentionally groups client processes +/// into buckets to keep Cloud Monitoring monitored resource target cardinality within quota limits. #[cfg(feature = "_experimental-builtin-metrics")] pub(crate) fn generate_client_hash(client_uid: &str) -> String { if client_uid.is_empty() { @@ -152,8 +173,111 @@ pub(crate) fn generate_client_hash(client_uid: &str) -> String { hash ^= byte as u64; hash = hash.wrapping_mul(0x100000001b3); } - let hash_24 = hash & 0xff_ffff; - format!("{hash_24:06x}") + let shifted = hash >> 54; + format!("{shifted:06x}") +} + +/// Parses a region name from a zone or region path/string. +/// For example: +/// - `"projects/12345/zones/us-central1-a"` -> `Some("us-central1")` +/// - `"projects/12345/regions/us-central1"` -> `Some("us-central1")` +/// - `"us-central1-a"` -> `Some("us-central1")` +/// - `"us-central1"` -> `Some("us-central1")` +#[cfg(feature = "_experimental-builtin-metrics")] +pub(crate) fn parse_region_from_zone_or_region(zone_or_region: &str) -> Option<&str> { + let trimmed = zone_or_region.trim().trim_end_matches('/'); + if trimmed.is_empty() { + return None; + } + let name = trimmed.rsplit('/').next()?; + if let Some((prefix, suffix)) = name.rsplit_once('-') + && suffix.len() == 1 + && prefix.contains('-') + { + return Some(prefix); + } + Some(name) +} + +#[cfg(feature = "_experimental-builtin-metrics")] +static DETECTED_LOCATION: OnceCell = OnceCell::const_new(); + +/// Detects the client's GCP location (e.g. `"us-central1"`), falling back to `"global"`. +/// +/// The result is cached process-wide in an [`OnceCell`] so the MDS query runs +/// at most once across the lifetime of the application. +#[cfg(feature = "_experimental-builtin-metrics")] +pub(crate) async fn detect_client_location(is_emulator: bool, is_plaintext: bool) -> String { + if is_emulator || is_plaintext { + return DEFAULT_CLIENT_LOCATION.to_string(); + } + + DETECTED_LOCATION + .get_or_init(resolve_client_location) + .await + .clone() +} + +/// Resolves the GCP location from environment variables or the Metadata Service. +#[cfg(feature = "_experimental-builtin-metrics")] +pub(crate) async fn resolve_client_location() -> String { + if let Ok(loc) = env::var("SPANNER_CLIENT_LOCATION") + && !loc.trim().is_empty() + { + return loc.trim().to_string(); + } + if let Ok(region) = env::var("GOOGLE_CLOUD_REGION") + && !region.trim().is_empty() + { + return region.trim().to_string(); + } + + fetch_location_from_mds() + .await + .unwrap_or_else(|| DEFAULT_CLIENT_LOCATION.to_string()) +} + +#[cfg(feature = "_experimental-builtin-metrics")] +async fn fetch_location_from_mds() -> Option { + let timeout_ms = env::var("SPANNER_CHECK_IS_RUNNING_ON_GCP_TIMEOUT") + .ok() + .and_then(|s| s.parse::().ok()) + .unwrap_or(DEFAULT_GCP_CHECK_TIMEOUT_MS); + let timeout_duration = Duration::from_millis(timeout_ms); + let connect_timeout_duration = + Duration::from_millis(timeout_ms.min(DEFAULT_GCP_CHECK_CONNECT_TIMEOUT_MS)); + + let host = + env::var(GCE_METADATA_HOST_ENV_VAR).unwrap_or_else(|_| DEFAULT_METADATA_ROOT.to_string()); + let base = if host.contains("://") { + host + } else { + format!("http://{host}") + }; + let base_url = Url::parse(base.trim_end_matches('/')).ok()?; + let url = base_url + .join(INSTANCE_ZONE_METADATA_PATH.trim_start_matches('/')) + .ok()?; + + let client = Client::builder() + .timeout(timeout_duration) + .connect_timeout(connect_timeout_duration) + .build() + .ok()?; + + let response = client + .get(url) + .header("Metadata-Flavor", "Google") + .send() + .await + .ok()?; + + if !response.status().is_success() { + return None; + } + + let text = response.text().await.ok()?; + parse_region_from_zone_or_region(&text).map(ToString::to_string) } /// Returns the library client identification string (`"spanner-rust/"`). @@ -195,7 +319,7 @@ impl Observability { database_name: &str, is_emulator: bool, ) -> Self { - let disable_builtin_metrics = std::env::var("SPANNER_DISABLE_BUILTIN_METRICS") + let disable_builtin_metrics = env::var("SPANNER_DISABLE_BUILTIN_METRICS") .map(|s| s.eq_ignore_ascii_case("true") || s == "1") .unwrap_or(false); let is_plaintext = config @@ -241,12 +365,13 @@ impl Observability { let client_uid = generate_client_uid(); let client_hash = generate_client_hash(&client_uid); let client_name = client_name(); + let location = detect_client_location(is_emulator, is_plaintext).await; let resource = Resource::builder() .with_attributes([ KeyValue::new("project_id", project_id.to_string()), KeyValue::new("instance_id", instance_id.to_string()), - KeyValue::new("location", "global"), + KeyValue::new("location", location), KeyValue::new("instance_config", "unknown"), KeyValue::new("client_hash", client_hash), ]) @@ -404,7 +529,7 @@ pub(crate) const AFE_SERVER_TIMING_HEADER: &str = "x-goog-spanner-enable-afe-ser #[cfg(feature = "_experimental-builtin-metrics")] static AFE_SERVER_TIMING_ENABLED: LazyLock = LazyLock::new(|| { - !std::env::var("SPANNER_DISABLE_AFE_SERVER_TIMING") + !env::var("SPANNER_DISABLE_AFE_SERVER_TIMING") .map(|val| val.eq_ignore_ascii_case("true") || val == "1") .unwrap_or(false) }); @@ -664,8 +789,11 @@ mod tests { use opentelemetry_sdk::metrics::InMemoryMetricExporter; use opentelemetry_sdk::metrics::PeriodicReader; use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData, ResourceMetrics}; + use scoped_env::ScopedEnv; + use serial_test::serial; use std::collections::HashMap; use std::fmt::Debug; + use tokio::net::TcpListener; #[test] fn traits() { @@ -881,7 +1009,6 @@ mod tests { assert_eq!(generate_client_hash(""), "000000"); let hash1 = generate_client_hash("test-client-uid"); - assert_eq!(hash1, "416874"); assert_eq!(hash1.len(), 6); assert!( hash1 @@ -890,10 +1017,282 @@ mod tests { "hash must be 6 lowercase hex characters, got {hash1}" ); + // Verify the 10-bit prefix limit (<= 0x3ff) + let val = u32::from_str_radix(&hash1, 16).expect("valid hex"); + assert!( + val <= 0x3ff, + "hash value {hash1} must fit in 10 bits (<= 0x3ff)" + ); + let hash2 = generate_client_hash("test-client-uid"); assert_eq!(hash1, hash2, "client hash must be deterministic"); + } + + /// Spawns an in-memory mock Google Compute Engine (GCE) Instance Metadata Server. + /// + /// The Metadata Server (often abbreviated as MDS) is a link-local HTTP service + /// (`http://169.254.169.254` or `http://metadata.google.internal`) available within Google Cloud + /// compute environments (such as Compute Engine VMs, Google Kubernetes Engine Pods, and Cloud Run). + /// Applications and client libraries query the Metadata Server to discover runtime metadata about + /// the host environment—including the current GCP zone and region, project ID, and temporary OAuth2 + /// service account access tokens—without requiring local credential files. + /// + /// In Cloud Spanner built-in metrics, the client queries the Metadata Server zone endpoint + /// (`/computeMetadata/v1/instance/zone`) upon startup to detect the client's geographic GCP region + /// (e.g., `us-central1` or `europe-west3`) and populate the `location` label of the + /// `spanner_instance_client` MonitoredResource. + /// + /// This test helper binds a local TCP listener on an ephemeral port, accepts incoming HTTP + /// requests in a loop, and matches each request URL against the caller-supplied `routes` + /// (defined as `(expected_path_substring, status_line, response_body)`). Any request path not + /// matching one of the supplied routes receives a standard `404 Not Found` response. + async fn spawn_mock_metadata_server( + routes: &[(&'static str, &'static str, &'static str)], + ) -> (String, String) { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind listener"); + let local_addr = listener.local_addr().expect("local_addr"); + let owned_routes: Vec<(&'static str, &'static str, &'static str)> = routes.to_vec(); + + tokio::spawn(async move { + loop { + let Ok((mut socket, _)) = listener.accept().await else { + break; + }; + let routes_for_task = owned_routes.clone(); + tokio::spawn(async move { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let mut buffer = [0u8; 2048]; + let Ok(bytes_read) = socket.read(&mut buffer).await else { + return; + }; + let request_string = String::from_utf8_lossy(&buffer[..bytes_read]); + + let mut matched_response = None; + for &(path, status_line, body) in &routes_for_task { + if request_string.contains(path) { + matched_response = Some(format!( + "HTTP/1.1 {status_line}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + )); + break; + } + } + + let response = matched_response.unwrap_or_else(|| { + "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + .to_string() + }); + let _ = socket.write_all(response.as_bytes()).await; + }); + } + }); + + (format!("http://{local_addr}"), local_addr.to_string()) + } - assert_eq!(generate_client_hash("spanner"), "727a8e"); + #[test] + fn parse_region_from_zone_or_region_cases() { + assert_eq!( + parse_region_from_zone_or_region("projects/12345/zones/us-central1-a"), + Some("us-central1") + ); + assert_eq!( + parse_region_from_zone_or_region("projects/12345/zones/us-central1-a/"), + Some("us-central1") + ); + assert_eq!( + parse_region_from_zone_or_region("projects/12345/zones/europe-west1-b"), + Some("europe-west1") + ); + assert_eq!( + parse_region_from_zone_or_region("projects/12345/zones/asia-northeast3-c"), + Some("asia-northeast3") + ); + assert_eq!( + parse_region_from_zone_or_region("projects/12345/regions/us-central1"), + Some("us-central1") + ); + assert_eq!( + parse_region_from_zone_or_region("projects/12345/regions/us-central1/"), + Some("us-central1") + ); + assert_eq!( + parse_region_from_zone_or_region("us-central1-f"), + Some("us-central1") + ); + assert_eq!( + parse_region_from_zone_or_region("us-central1"), + Some("us-central1") + ); + assert_eq!( + parse_region_from_zone_or_region("us-central1/"), + Some("us-central1") + ); + assert_eq!( + parse_region_from_zone_or_region("projects/12345/zones/us-central1-a\n"), + Some("us-central1") + ); + assert_eq!( + parse_region_from_zone_or_region("projects/12345/zones/us-central1-a\r\n"), + Some("us-central1") + ); + assert_eq!(parse_region_from_zone_or_region("global"), Some("global")); + assert_eq!(parse_region_from_zone_or_region(""), None); + assert_eq!(parse_region_from_zone_or_region(" "), None); + assert_eq!(parse_region_from_zone_or_region("/"), None); + assert_eq!(parse_region_from_zone_or_region("///"), None); + } + + #[tokio::test] + #[serial] + async fn resolve_client_location_env_overrides() { + // Explicit SPANNER_CLIENT_LOCATION override + { + let _env = ScopedEnv::set("SPANNER_CLIENT_LOCATION", "europe-west4"); + assert_eq!(resolve_client_location().await, "europe-west4"); + } + + // GOOGLE_CLOUD_REGION override + { + let _env = ScopedEnv::set("GOOGLE_CLOUD_REGION", "us-east1"); + assert_eq!(resolve_client_location().await, "us-east1"); + } + + // SPANNER_CLIENT_LOCATION takes precedence over GOOGLE_CLOUD_REGION + { + let _env_loc = ScopedEnv::set("SPANNER_CLIENT_LOCATION", "europe-west4"); + let _env_reg = ScopedEnv::set("GOOGLE_CLOUD_REGION", "us-east1"); + assert_eq!( + resolve_client_location().await, + "europe-west4", + "SPANNER_CLIENT_LOCATION must take precedence over GOOGLE_CLOUD_REGION" + ); + } + + // Whitespace-only values must be ignored and fall back to MDS / global + { + let _env_loc = ScopedEnv::set("SPANNER_CLIENT_LOCATION", " "); + let _env_reg = ScopedEnv::set("GOOGLE_CLOUD_REGION", " "); + let _env_host = ScopedEnv::set("GCE_METADATA_HOST", "http://invalid-host:1"); + let _env_timeout = ScopedEnv::set("SPANNER_CHECK_IS_RUNNING_ON_GCP_TIMEOUT", "50"); + assert_eq!(resolve_client_location().await, "global"); + } + } + + #[tokio::test] + #[serial] + async fn resolve_client_location_with_mock_metadata_server() { + let (url, _addr) = spawn_mock_metadata_server(&[ + ( + INSTANCE_ZONE_METADATA_PATH, + "200 OK", + "projects/123/zones/us-west1-b", + ), + ( + "/computeMetadata/v1/universe/universe_domain", + "200 OK", + "googleapis.com", + ), + ]) + .await; + + let _env_host = ScopedEnv::set("GCE_METADATA_HOST", &url); + let _env_timeout = ScopedEnv::set("SPANNER_CHECK_IS_RUNNING_ON_GCP_TIMEOUT", "1000"); + let _env_loc = ScopedEnv::remove("SPANNER_CLIENT_LOCATION"); + let _env_reg = ScopedEnv::remove("GOOGLE_CLOUD_REGION"); + + let location = resolve_client_location().await; + assert_eq!(location, "us-west1"); + } + + #[tokio::test] + #[serial] + async fn resolve_client_location_with_mock_metadata_server_without_scheme_and_invalid_timeout() + { + let (_url, addr) = spawn_mock_metadata_server(&[ + ( + INSTANCE_ZONE_METADATA_PATH, + "200 OK", + "projects/456/zones/europe-west3-c", + ), + ( + "/computeMetadata/v1/universe/universe_domain", + "200 OK", + "googleapis.com", + ), + ]) + .await; + + // Test GCE_METADATA_HOST without "http://" prefix (e.g. "127.0.0.1:12345") + let _env_host = ScopedEnv::set("GCE_METADATA_HOST", &addr); + // Test invalid timeout string falling back to default timeout + let _env_timeout = + ScopedEnv::set("SPANNER_CHECK_IS_RUNNING_ON_GCP_TIMEOUT", "not-a-number"); + let _env_loc = ScopedEnv::remove("SPANNER_CLIENT_LOCATION"); + let _env_reg = ScopedEnv::remove("GOOGLE_CLOUD_REGION"); + + let location = resolve_client_location().await; + assert_eq!(location, "europe-west3"); + } + + #[tokio::test] + #[serial] + async fn resolve_client_location_with_mock_metadata_server_error_status() { + let (url, _addr) = spawn_mock_metadata_server(&[ + ( + INSTANCE_ZONE_METADATA_PATH, + "500 Internal Server Error", + "Internal Server Error", + ), + ( + "/computeMetadata/v1/universe/universe_domain", + "200 OK", + "googleapis.com", + ), + ]) + .await; + + let _env_host = ScopedEnv::set("GCE_METADATA_HOST", &url); + let _env_timeout = ScopedEnv::set("SPANNER_CHECK_IS_RUNNING_ON_GCP_TIMEOUT", "1000"); + let _env_loc = ScopedEnv::remove("SPANNER_CLIENT_LOCATION"); + let _env_reg = ScopedEnv::remove("GOOGLE_CLOUD_REGION"); + + let location = resolve_client_location().await; + assert_eq!( + location, "global", + "HTTP 500 error from metadata server must fall back to 'global'" + ); + } + + #[tokio::test] + #[serial] + async fn resolve_client_location_unreachable_metadata_server_fallback() { + // Test connection failure / unreachable host falling back to global + let _env_host = ScopedEnv::set("GCE_METADATA_HOST", "http://127.0.0.1:1"); + let _env_timeout = ScopedEnv::set("SPANNER_CHECK_IS_RUNNING_ON_GCP_TIMEOUT", "50"); + let _env_loc = ScopedEnv::remove("SPANNER_CLIENT_LOCATION"); + let _env_reg = ScopedEnv::remove("GOOGLE_CLOUD_REGION"); + + let location = resolve_client_location().await; + assert_eq!( + location, "global", + "unreachable metadata server host must fall back to 'global'" + ); + } + + #[tokio::test] + #[serial] + async fn detect_client_location_caching() { + // Emulator bypasses MDS and caching + assert_eq!(detect_client_location(true, false).await, "global"); + // Plaintext bypasses MDS and caching + assert_eq!(detect_client_location(false, true).await, "global"); + + // Executes OnceCell caching path + let loc = detect_client_location(false, false).await; + assert!(!loc.is_empty(), "detected location must not be empty"); } #[test]