GeezerVison is showing github.com — fix(metrics): prevent thread leak by ensuring singleton initializatio… · googleapis/python-spanner@e792136
This repository was archived by the owner on Jun 8, 2026. It is now read-only.
File treeExpand file treeCollapse file tree Open diff view settings Expand file treeCollapse file tree Open diff view settings Collapse file Copy file name to clipboardExpand all lines: .kokoro/presubmit/presubmit.cfg +1-1Lines changed: 1 addition & 1 deletion | Original file line number | Diff line number | Diff line change |
|---|
| 3 | 3 | # Only run a subset of all nox sessions | 4 | 4 | env_vars: { | 5 | 5 | key: "NOX_SESSION" | 6 | | - value: "unit-3.9 unit-3.12 cover docs docfx" | | 6 | + value: "unit-3.10 unit-3.12 cover docs docfx" | 7 | 7 | } |
Collapse file Copy file name to clipboardExpand all lines: google/cloud/spanner_v1/batch.py +1Lines changed: 1 addition & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
| 13 | 13 | # limitations under the License. | 14 | 14 |
| 15 | 15 | """Context manager for Cloud Spanner batched writes.""" | | 16 | + | 16 | 17 | import functools | 17 | 18 | from typing import List, Optional | 18 | 19 |
| |
Collapse file Copy file name to clipboardExpand all lines: google/cloud/spanner_v1/client.py +42-20Lines changed: 42 additions & 20 deletions | Original file line number | Diff line number | Diff line change |
|---|
| 23 | 23 | * a :class:`~google.cloud.spanner_v1.instance.Instance` owns a | 24 | 24 | :class:`~google.cloud.spanner_v1.database.Database` | 25 | 25 | """ | | 26 | + | 26 | 27 | import grpc | 27 | 28 | import os | 28 | 29 | import logging | 29 | 30 | import warnings | | 31 | +import threading | 30 | 32 |
| 31 | 33 | from google.api_core.gapic_v1 import client_info | 32 | 34 | from google.auth.credentials import AnonymousCredentials | @@ -99,11 +101,50 @@ def _get_spanner_optimizer_statistics_package(): | 99 | 101 |
| 100 | 102 | log = logging.getLogger(__name__) | 101 | 103 |
| | 104 | +_metrics_monitor_initialized = False | | 105 | +_metrics_monitor_lock = threading.Lock() | | 106 | + | 102 | 107 |
| 103 | 108 | def _get_spanner_enable_builtin_metrics_env(): | 104 | 109 | return os.getenv(SPANNER_DISABLE_BUILTIN_METRICS_ENV_VAR) != "true" | 105 | 110 |
| 106 | 111 |
| | 112 | +def _initialize_metrics(project, credentials): | | 113 | + """ | | 114 | + Initializes the Spanner built-in metrics. | | 115 | +
| | 116 | + This function sets up the OpenTelemetry MeterProvider and the SpannerMetricsTracerFactory. | | 117 | + It uses a lock to ensure that initialization happens only once. | | 118 | + """ | | 119 | + global _metrics_monitor_initialized | | 120 | + if not _metrics_monitor_initialized: | | 121 | + with _metrics_monitor_lock: | | 122 | + if not _metrics_monitor_initialized: | | 123 | + meter_provider = metrics.NoOpMeterProvider() | | 124 | + try: | | 125 | + if not _get_spanner_emulator_host(): | | 126 | + meter_provider = MeterProvider( | | 127 | + metric_readers=[ | | 128 | + PeriodicExportingMetricReader( | | 129 | + CloudMonitoringMetricsExporter( | | 130 | + project_id=project, | | 131 | + credentials=credentials, | | 132 | + ), | | 133 | + export_interval_millis=METRIC_EXPORT_INTERVAL_MS, | | 134 | + ), | | 135 | + ] | | 136 | + ) | | 137 | + metrics.set_meter_provider(meter_provider) | | 138 | + SpannerMetricsTracerFactory() | | 139 | + _metrics_monitor_initialized = True | | 140 | + except Exception as e: | | 141 | + # log is already defined at module level | | 142 | + log.warning( | | 143 | + "Failed to initialize Spanner built-in metrics. Error: %s", | | 144 | + e, | | 145 | + ) | | 146 | + | | 147 | + | 107 | 148 | class Client(ClientWithProject): | 108 | 149 | """Client for interacting with Cloud Spanner API. | 109 | 150 |
| @@ -251,31 +292,12 @@ def __init__( | 251 | 292 | "http://" in self._emulator_host or "https://" in self._emulator_host | 252 | 293 | ): | 253 | 294 | warnings.warn(_EMULATOR_HOST_HTTP_SCHEME) | 254 | | - # Check flag to enable Spanner builtin metrics | 255 | 295 | if ( | 256 | 296 | _get_spanner_enable_builtin_metrics_env() | 257 | 297 | and not disable_builtin_metrics | 258 | 298 | and HAS_GOOGLE_CLOUD_MONITORING_INSTALLED | 259 | 299 | ): | 260 | | - meter_provider = metrics.NoOpMeterProvider() | 261 | | - try: | 262 | | - if not _get_spanner_emulator_host(): | 263 | | - meter_provider = MeterProvider( | 264 | | - metric_readers=[ | 265 | | - PeriodicExportingMetricReader( | 266 | | - CloudMonitoringMetricsExporter( | 267 | | - project_id=project, credentials=credentials | 268 | | - ), | 269 | | - export_interval_millis=METRIC_EXPORT_INTERVAL_MS, | 270 | | - ), | 271 | | - ] | 272 | | - ) | 273 | | - metrics.set_meter_provider(meter_provider) | 274 | | - SpannerMetricsTracerFactory() | 275 | | - except Exception as e: | 276 | | - log.warning( | 277 | | - "Failed to initialize Spanner built-in metrics. Error: %s", e | 278 | | - ) | | 300 | + _initialize_metrics(project, credentials) | 279 | 301 | else: | 280 | 302 | SpannerMetricsTracerFactory(enabled=False) | 281 | 303 |
| |
Collapse file Copy file name to clipboardExpand all lines: google/cloud/spanner_v1/metrics/metrics_capture.py +17-7Lines changed: 17 additions & 7 deletions | Original file line number | Diff line number | Diff line change |
|---|
| 20 | 20 | performance monitoring. | 21 | 21 | """ | 22 | 22 |
| | 23 | +from contextvars import Token | | 24 | + | 23 | 25 | from .spanner_metrics_tracer_factory import SpannerMetricsTracerFactory | 24 | 26 |
| 25 | 27 |
| @@ -30,6 +32,9 @@ class MetricsCapture: | 30 | 32 | the start and completion of metrics tracing for a given operation. | 31 | 33 | """ | 32 | 34 |
| | 35 | + _token: Token | | 36 | + """Token to reset the context variable after the operation completes.""" | | 37 | + | 33 | 38 | def __enter__(self): | 34 | 39 | """Enter the runtime context related to this object. | 35 | 40 |
| @@ -45,11 +50,11 @@ def __enter__(self): | 45 | 50 | return self | 46 | 51 |
| 47 | 52 | # Define a new metrics tracer for the new operation | 48 | | - SpannerMetricsTracerFactory.current_metrics_tracer = ( | 49 | | - factory.create_metrics_tracer() | 50 | | - ) | 51 | | - if SpannerMetricsTracerFactory.current_metrics_tracer: | 52 | | - SpannerMetricsTracerFactory.current_metrics_tracer.record_operation_start() | | 53 | + # Set the context var and keep the token for reset | | 54 | + tracer = factory.create_metrics_tracer() | | 55 | + self._token = SpannerMetricsTracerFactory.set_current_tracer(tracer) | | 56 | + if tracer: | | 57 | + tracer.record_operation_start() | 53 | 58 | return self | 54 | 59 |
| 55 | 60 | def __exit__(self, exc_type, exc_value, traceback): | @@ -70,6 +75,11 @@ def __exit__(self, exc_type, exc_value, traceback): | 70 | 75 | if not SpannerMetricsTracerFactory().enabled: | 71 | 76 | return False | 72 | 77 |
| 73 | | - if SpannerMetricsTracerFactory.current_metrics_tracer: | 74 | | - SpannerMetricsTracerFactory.current_metrics_tracer.record_operation_completion() | | 78 | + tracer = SpannerMetricsTracerFactory.get_current_tracer() | | 79 | + if tracer: | | 80 | + tracer.record_operation_completion() | | 81 | + | | 82 | + # Reset the context var using the token | | 83 | + if getattr(self, "_token", None): | | 84 | + SpannerMetricsTracerFactory.reset_current_tracer(self._token) | 75 | 85 | return False # Propagate the exception if any |
Collapse file Copy file name to clipboardExpand all lines: google/cloud/spanner_v1/metrics/metrics_interceptor.py +20-24Lines changed: 20 additions & 24 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -97,22 +97,17 @@ def _set_metrics_tracer_attributes(self, resources: Dict[str, str]) -> None: | 97 | 97 | Args: | 98 | 98 | resources (Dict[str, str]): A dictionary containing project, instance, and database information. | 99 | 99 | """ | 100 | | - if SpannerMetricsTracerFactory.current_metrics_tracer is None: | | 100 | + tracer = SpannerMetricsTracerFactory.get_current_tracer() | | 101 | + if tracer is None: | 101 | 102 | return | 102 | 103 |
| 103 | 104 | if resources: | 104 | 105 | if "project" in resources: | 105 | | - SpannerMetricsTracerFactory.current_metrics_tracer.set_project( | 106 | | - resources["project"] | 107 | | - ) | | 106 | + tracer.set_project(resources["project"]) | 108 | 107 | if "instance" in resources: | 109 | | - SpannerMetricsTracerFactory.current_metrics_tracer.set_instance( | 110 | | - resources["instance"] | 111 | | - ) | | 108 | + tracer.set_instance(resources["instance"]) | 112 | 109 | if "database" in resources: | 113 | | - SpannerMetricsTracerFactory.current_metrics_tracer.set_database( | 114 | | - resources["database"] | 115 | | - ) | | 110 | + tracer.set_database(resources["database"]) | 116 | 111 |
| 117 | 112 | def intercept(self, invoked_method, request_or_iterator, call_details): | 118 | 113 | """Intercept gRPC calls to collect metrics. | @@ -126,31 +121,32 @@ def intercept(self, invoked_method, request_or_iterator, call_details): | 126 | 121 | The RPC response | 127 | 122 | """ | 128 | 123 | factory = SpannerMetricsTracerFactory() | 129 | | - if ( | 130 | | - SpannerMetricsTracerFactory.current_metrics_tracer is None | 131 | | - or not factory.enabled | 132 | | - ): | | 124 | + tracer = SpannerMetricsTracerFactory.get_current_tracer() | | 125 | + if tracer is None or not factory.enabled: | 133 | 126 | return invoked_method(request_or_iterator, call_details) | 134 | 127 |
| 135 | 128 | # Setup Metric Tracer attributes from call details | 136 | | - ## Extract Project / Instance / Databse from header information | 137 | | - resources = self._extract_resource_from_path(call_details.metadata) | 138 | | - self._set_metrics_tracer_attributes(resources) | | 129 | + ## Extract Project / Instance / Database from header information if not already set | | 130 | + if not ( | | 131 | + tracer.client_attributes.get("project_id") | | 132 | + and tracer.client_attributes.get("instance_id") | | 133 | + and tracer.client_attributes.get("database") | | 134 | + ): | | 135 | + resources = self._extract_resource_from_path(call_details.metadata) | | 136 | + self._set_metrics_tracer_attributes(resources) | 139 | 137 |
| 140 | 138 | ## Format method to be be spanner.<method name> | 141 | 139 | method_name = self._remove_prefix( | 142 | 140 | call_details.method, SPANNER_METHOD_PREFIX | 143 | 141 | ).replace("/", ".") | 144 | 142 |
| 145 | | - SpannerMetricsTracerFactory.current_metrics_tracer.set_method(method_name) | 146 | | - SpannerMetricsTracerFactory.current_metrics_tracer.record_attempt_start() | | 143 | + tracer.set_method(method_name) | | 144 | + tracer.record_attempt_start() | 147 | 145 | response = invoked_method(request_or_iterator, call_details) | 148 | | - SpannerMetricsTracerFactory.current_metrics_tracer.record_attempt_completion() | | 146 | + tracer.record_attempt_completion() | 149 | 147 |
| 150 | 148 | # Process and send GFE metrics if enabled | 151 | | - if SpannerMetricsTracerFactory.current_metrics_tracer.gfe_enabled: | | 149 | + if tracer.gfe_enabled: | 152 | 150 | metadata = response.initial_metadata() | 153 | | - SpannerMetricsTracerFactory.current_metrics_trace.record_gfe_metrics( | 154 | | - metadata | 155 | | - ) | | 151 | + tracer.record_gfe_metrics(metadata) | 156 | 152 | return response |
Collapse file Copy file name to clipboardExpand all lines: google/cloud/spanner_v1/metrics/spanner_metrics_tracer_factory.py +17-2Lines changed: 17 additions & 2 deletions | Original file line number | Diff line number | Diff line change |
|---|
| 19 | 19 | import os | 20 | 20 | import logging | 21 | 21 | from .constants import SPANNER_SERVICE_NAME | | 22 | +import contextvars | 22 | 23 |
| 23 | 24 | try: | 24 | 25 | import mmh3 | @@ -43,7 +44,9 @@ class SpannerMetricsTracerFactory(MetricsTracerFactory): | 43 | 44 | """A factory for creating SpannerMetricsTracer instances.""" | 44 | 45 |
| 45 | 46 | _metrics_tracer_factory: "SpannerMetricsTracerFactory" = None | 46 | | - current_metrics_tracer: MetricsTracer = None | | 47 | + _current_metrics_tracer_ctx = contextvars.ContextVar( | | 48 | + "current_metrics_tracer", default=None | | 49 | + ) | 47 | 50 |
| 48 | 51 | def __new__( | 49 | 52 | cls, enabled: bool = True, gfe_enabled: bool = False | @@ -80,10 +83,22 @@ def __new__( | 80 | 83 | cls._metrics_tracer_factory.gfe_enabled = gfe_enabled | 81 | 84 |
| 82 | 85 | if cls._metrics_tracer_factory.enabled != enabled: | 83 | | - cls._metrics_tracer_factory.enabeld = enabled | | 86 | + cls._metrics_tracer_factory.enabled = enabled | 84 | 87 |
| 85 | 88 | return cls._metrics_tracer_factory | 86 | 89 |
| | 90 | + @staticmethod | | 91 | + def get_current_tracer() -> MetricsTracer: | | 92 | + return SpannerMetricsTracerFactory._current_metrics_tracer_ctx.get() | | 93 | + | | 94 | + @staticmethod | | 95 | + def set_current_tracer(tracer: MetricsTracer) -> contextvars.Token: | | 96 | + return SpannerMetricsTracerFactory._current_metrics_tracer_ctx.set(tracer) | | 97 | + | | 98 | + @staticmethod | | 99 | + def reset_current_tracer(token: contextvars.Token): | | 100 | + SpannerMetricsTracerFactory._current_metrics_tracer_ctx.reset(token) | | 101 | + | 87 | 102 | @staticmethod | 88 | 103 | def _generate_client_uid() -> str: | 89 | 104 | """Generate a client UID in the form of uuidv4@pid@hostname. | |
Collapse file Copy file name to clipboard +27Lines changed: 27 additions & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
| | 1 | +# Copyright 2025 Google LLC | | 2 | +# | | 3 | +# Licensed under the Apache License, Version 2.0 (the "License"); | | 4 | +# you may not use this file except in compliance with the License. | | 5 | +# You may obtain a copy of the License at | | 6 | +# | | 7 | +# http://www.apache.org/licenses/LICENSE-2.0 | | 8 | +# | | 9 | +# Unless required by applicable law or agreed to in writing, software | | 10 | +# distributed under the License is distributed on an "AS IS" BASIS, | | 11 | +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | | 12 | +# See the License for the specific language governing permissions and | | 13 | +# limitations under the License. | | 14 | + | | 15 | +import pytest | | 16 | +from unittest.mock import patch | | 17 | + | | 18 | + | | 19 | +@pytest.fixture(autouse=True) | | 20 | +def mock_periodic_exporting_metric_reader(): | | 21 | + """Globally mock PeriodicExportingMetricReader to prevent real network calls.""" | | 22 | + with patch( | | 23 | + "google.cloud.spanner_v1.client.PeriodicExportingMetricReader" | | 24 | + ) as mock_client_reader, patch( | | 25 | + "opentelemetry.sdk.metrics.export.PeriodicExportingMetricReader" | | 26 | + ): | | 27 | + yield mock_client_reader |
|