fix(metrics): prevent thread leak by ensuring singleton initialization (#1492)

****Summary:****
This PR fixes a critical memory and thread leak in the
google-cloud-spanner client when built-in metrics are enabled (default
behavior).
Previously, the Client constructor unconditionally initialized a new
OpenTelemetry MeterProvider and PeriodicExportingMetricReader on every
instantiation. Each reader spawned a new background thread for metric
exporting that was never cleaned up or reused. In environments where
Client objects are frequently created (e.g., Cloud Functions, web
servers, or data pipelines), this caused a linear accumulation of
threads, leading to RuntimeError: can't start new thread and OOM
crashes.
****Fix Implementation:****
***Refactored Metrics Initialization (Thread Safety & Memory Leak
Fix)***:
Implemented a Singleton pattern for the OpenTelemetry MeterProvider
using threading.Lock to prevent infinite background thread creation
(memory leak).
Moved metrics initialization logic to a cleaner helper function
_initialize_metrics in client.py.
Replaced global mutable state in SpannerMetricsTracerFactory with
contextvars.ContextVar to ensure thread-safe, isolated metric tracing
across concurrent requests.
Updated MetricsInterceptor and MetricsCapture to correctly use the
thread-local tracer.
***Fixed Batch.commit Idempotency (AlreadyExists Regression):***
Modified Batch.commit to initialize nth_request and the attempt counter
outside the retry loop.
This ensures that retries (e.g., on ABORTED) reuse the same Request ID,
allowing Cloud Spanner to correctly deduplicate requests and preventing
spurious AlreadyExists (409) errors.
***Verification:***
Added tests/unit/test_metrics_concurrency.py to verify tracer isolation
and thread safety.
Cleaned up tests/unit/test_metrics.py and consolidated mocks in
conftest.py.

Author: sinhasubham

.kokoro/presubmit/presubmit.cfg

@@ -3,5 +3,5 @@
 # Only run a subset of all nox sessions
 env_vars: {
  key: "NOX_SESSION"
- value: "unit-3.9 unit-3.12 cover docs docfx"
+ value: "unit-3.10 unit-3.12 cover docs docfx"
 }

google/cloud/spanner_v1/batch.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 """Context manager for Cloud Spanner batched writes."""
+
 importfunctools
 fromtypingimportList, Optional
 

google/cloud/spanner_v1/client.py

@@ -23,10 +23,12 @@
 * a :class:`~google.cloud.spanner_v1.instance.Instance` owns a
  :class:`~google.cloud.spanner_v1.database.Database`
 """
+
 importgrpc
 importos
 importlogging
 importwarnings
+importthreading
 
 fromgoogle.api_core.gapic_v1importclient_info
 fromgoogle.auth.credentialsimportAnonymousCredentials
@@ -99,11 +101,50 @@ def _get_spanner_optimizer_statistics_package():
 
 log=logging.getLogger(__name__)
 
+_metrics_monitor_initialized=False
+_metrics_monitor_lock=threading.Lock()
+
 
 def_get_spanner_enable_builtin_metrics_env():
 returnos.getenv(SPANNER_DISABLE_BUILTIN_METRICS_ENV_VAR) !="true"
 
 
+def_initialize_metrics(project, credentials):
+"""
+ Initializes the Spanner built-in metrics.
+
+ This function sets up the OpenTelemetry MeterProvider and the SpannerMetricsTracerFactory.
+ It uses a lock to ensure that initialization happens only once.
+ """
+global_metrics_monitor_initialized
+ifnot_metrics_monitor_initialized:
+with_metrics_monitor_lock:
+ifnot_metrics_monitor_initialized:
+meter_provider=metrics.NoOpMeterProvider()
+try:
+ifnot_get_spanner_emulator_host():
+meter_provider=MeterProvider(
+metric_readers=[
+PeriodicExportingMetricReader(
+CloudMonitoringMetricsExporter(
+project_id=project,
+credentials=credentials,
+ ),
+export_interval_millis=METRIC_EXPORT_INTERVAL_MS,
+ ),
+ ]
+ )
+metrics.set_meter_provider(meter_provider)
+SpannerMetricsTracerFactory()
+_metrics_monitor_initialized=True
+exceptExceptionase:
+# log is already defined at module level
+log.warning(
+"Failed to initialize Spanner built-in metrics. Error: %s",
+e,
+ )
+
+
 classClient(ClientWithProject):
 """Client for interacting with Cloud Spanner API.
 
@@ -251,31 +292,12 @@ def __init__(
 "http://"inself._emulator_hostor"https://"inself._emulator_host
  ):
 warnings.warn(_EMULATOR_HOST_HTTP_SCHEME)
-# Check flag to enable Spanner builtin metrics
 if (
 _get_spanner_enable_builtin_metrics_env()
 andnotdisable_builtin_metrics
 andHAS_GOOGLE_CLOUD_MONITORING_INSTALLED
  ):
-meter_provider=metrics.NoOpMeterProvider()
-try:
-ifnot_get_spanner_emulator_host():
-meter_provider=MeterProvider(
-metric_readers=[
-PeriodicExportingMetricReader(
-CloudMonitoringMetricsExporter(
-project_id=project, credentials=credentials
- ),
-export_interval_millis=METRIC_EXPORT_INTERVAL_MS,
- ),
- ]
- )
-metrics.set_meter_provider(meter_provider)
-SpannerMetricsTracerFactory()
-exceptExceptionase:
-log.warning(
-"Failed to initialize Spanner built-in metrics. Error: %s", e
- )
+_initialize_metrics(project, credentials)
 else:
 SpannerMetricsTracerFactory(enabled=False)
 

google/cloud/spanner_v1/metrics/metrics_capture.py

@@ -20,6 +20,8 @@
 performance monitoring.
 """
 
+fromcontextvarsimportToken
+
 from .spanner_metrics_tracer_factoryimportSpannerMetricsTracerFactory
 
 
@@ -30,6 +32,9 @@ class MetricsCapture:
  the start and completion of metrics tracing for a given operation.
  """
 
+_token: Token
+"""Token to reset the context variable after the operation completes."""
+
 def__enter__(self):
 """Enter the runtime context related to this object.
 
@@ -45,11 +50,11 @@ def __enter__(self):
 returnself
 
 # Define a new metrics tracer for the new operation
-SpannerMetricsTracerFactory.current_metrics_tracer= (
-factory.create_metrics_tracer()
- )
-ifSpannerMetricsTracerFactory.current_metrics_tracer:
-SpannerMetricsTracerFactory.current_metrics_tracer.record_operation_start()
+# Set the context var and keep the token for reset
+tracer=factory.create_metrics_tracer()
+self._token=SpannerMetricsTracerFactory.set_current_tracer(tracer)
+iftracer:
+tracer.record_operation_start()
 returnself
 
 def__exit__(self, exc_type, exc_value, traceback):
@@ -70,6 +75,11 @@ def __exit__(self, exc_type, exc_value, traceback):
 ifnotSpannerMetricsTracerFactory().enabled:
 returnFalse
 
-ifSpannerMetricsTracerFactory.current_metrics_tracer:
-SpannerMetricsTracerFactory.current_metrics_tracer.record_operation_completion()
+tracer=SpannerMetricsTracerFactory.get_current_tracer()
+iftracer:
+tracer.record_operation_completion()
+
+# Reset the context var using the token
+ifgetattr(self, "_token", None):
+SpannerMetricsTracerFactory.reset_current_tracer(self._token)
 returnFalse# Propagate the exception if any

google/cloud/spanner_v1/metrics/metrics_interceptor.py

@@ -97,22 +97,17 @@ def _set_metrics_tracer_attributes(self, resources: Dict[str, str]) -> None:
  Args:
  resources (Dict[str, str]): A dictionary containing project, instance, and database information.
  """
-ifSpannerMetricsTracerFactory.current_metrics_tracerisNone:
+tracer=SpannerMetricsTracerFactory.get_current_tracer()
+iftracerisNone:
 return
 
 ifresources:
 if"project"inresources:
-SpannerMetricsTracerFactory.current_metrics_tracer.set_project(
-resources["project"]
- )
+tracer.set_project(resources["project"])
 if"instance"inresources:
-SpannerMetricsTracerFactory.current_metrics_tracer.set_instance(
-resources["instance"]
- )
+tracer.set_instance(resources["instance"])
 if"database"inresources:
-SpannerMetricsTracerFactory.current_metrics_tracer.set_database(
-resources["database"]
- )
+tracer.set_database(resources["database"])
 
 defintercept(self, invoked_method, request_or_iterator, call_details):
 """Intercept gRPC calls to collect metrics.
@@ -126,31 +121,32 @@ def intercept(self, invoked_method, request_or_iterator, call_details):
  The RPC response
  """
 factory=SpannerMetricsTracerFactory()
-if (
-SpannerMetricsTracerFactory.current_metrics_tracerisNone
-ornotfactory.enabled
- ):
+tracer=SpannerMetricsTracerFactory.get_current_tracer()
+iftracerisNoneornotfactory.enabled:
 returninvoked_method(request_or_iterator, call_details)
 
 # Setup Metric Tracer attributes from call details
-## Extract Project / Instance / Databse from header information
-resources=self._extract_resource_from_path(call_details.metadata)
-self._set_metrics_tracer_attributes(resources)
+## Extract Project / Instance / Database from header information if not already set
+ifnot (
+tracer.client_attributes.get("project_id")
+andtracer.client_attributes.get("instance_id")
+andtracer.client_attributes.get("database")
+ ):
+resources=self._extract_resource_from_path(call_details.metadata)
+self._set_metrics_tracer_attributes(resources)
 
 ## Format method to be be spanner.<method name>
 method_name=self._remove_prefix(
 call_details.method, SPANNER_METHOD_PREFIX
  ).replace("/", ".")
 
-SpannerMetricsTracerFactory.current_metrics_tracer.set_method(method_name)
-SpannerMetricsTracerFactory.current_metrics_tracer.record_attempt_start()
+tracer.set_method(method_name)
+tracer.record_attempt_start()
 response=invoked_method(request_or_iterator, call_details)
-SpannerMetricsTracerFactory.current_metrics_tracer.record_attempt_completion()
+tracer.record_attempt_completion()
 
 # Process and send GFE metrics if enabled
-ifSpannerMetricsTracerFactory.current_metrics_tracer.gfe_enabled:
+iftracer.gfe_enabled:
 metadata=response.initial_metadata()
-SpannerMetricsTracerFactory.current_metrics_trace.record_gfe_metrics(
-metadata
- )
+tracer.record_gfe_metrics(metadata)
 returnresponse

google/cloud/spanner_v1/metrics/spanner_metrics_tracer_factory.py

@@ -19,6 +19,7 @@
 importos
 importlogging
 from .constantsimportSPANNER_SERVICE_NAME
+importcontextvars
 
 try:
 importmmh3
@@ -43,7 +44,9 @@ class SpannerMetricsTracerFactory(MetricsTracerFactory):
 """A factory for creating SpannerMetricsTracer instances."""
 
 _metrics_tracer_factory: "SpannerMetricsTracerFactory"=None
-current_metrics_tracer: MetricsTracer=None
+_current_metrics_tracer_ctx=contextvars.ContextVar(
+"current_metrics_tracer", default=None
+ )
 
 def__new__(
 cls, enabled: bool=True, gfe_enabled: bool=False
@@ -80,10 +83,22 @@ def __new__(
 cls._metrics_tracer_factory.gfe_enabled=gfe_enabled
 
 ifcls._metrics_tracer_factory.enabled!=enabled:
-cls._metrics_tracer_factory.enabeld=enabled
+cls._metrics_tracer_factory.enabled=enabled
 
 returncls._metrics_tracer_factory
 
+@staticmethod
+defget_current_tracer() ->MetricsTracer:
+returnSpannerMetricsTracerFactory._current_metrics_tracer_ctx.get()
+
+@staticmethod
+defset_current_tracer(tracer: MetricsTracer) ->contextvars.Token:
+returnSpannerMetricsTracerFactory._current_metrics_tracer_ctx.set(tracer)
+
+@staticmethod
+defreset_current_tracer(token: contextvars.Token):
+SpannerMetricsTracerFactory._current_metrics_tracer_ctx.reset(token)
+
 @staticmethod
 def_generate_client_uid() ->str:
 """Generate a client UID in the form of uuidv4@pid@hostname.

tests/unit/conftest.py

@@ -0,0 +1,27 @@
+# Copyright 2025 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
+#
+# http://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.
+
+importpytest
+fromunittest.mockimportpatch
+
+
+@pytest.fixture(autouse=True)
+defmock_periodic_exporting_metric_reader():
+"""Globally mock PeriodicExportingMetricReader to prevent real network calls."""
+withpatch(
+"google.cloud.spanner_v1.client.PeriodicExportingMetricReader"
+ ) asmock_client_reader, patch(
+"opentelemetry.sdk.metrics.export.PeriodicExportingMetricReader"
+ ):
+yieldmock_client_reader