Repository navigation
feat(storage): add OpenTelemetry standard metrics for transfer_manager - #18945
shradhakatyal wants to merge 12 commits into
Conversation
…d helpers Implement core OpenTelemetry standard metrics infrastructure for GCS: - Add _MetricsRecorder with the 9 standard metric instruments - Add custom latency (165 buckets) and byte-size (45 buckets) histogram views - Add classify_error_type for low-cardinality error.type mapping - Add normalize_url_template for bounded HTTP url.template routing - Add start_operation ContextVar for outermost logical operation tracking and nested call suppression [Generated-by: AI]
…branch coverage [Generated-by: AI]
…ncellations [Generated-by: AI]
- Add gcp.storage.client.retries counter and rename TTFB metric to gcp.storage.client.ttfb - Remove gcp.client.repo from _COMMON_ATTRIBUTES - Update _build_size_buckets to generate 48 buckets matching Go and Java SDKs - Align per-metric attributes across logical operation, RPC/HTTP attempt, error, retry, and TTFB recorders [Generated-by: AI]
…fy CONNECTION_ERROR - Scope histogram View registrations to meter_name="google.cloud.storage" - Guard normalize_url_template and _extract_server_address against malformed URIs and gRPC authority schemes - Consolidate CONNECTION_REFUSED, CONNECTION_RESET, and CONNECTIVITY error labels into CONNECTION_ERROR [Generated-by: AI]
…port - Wire _MetricsRecorder lifecycle, enable_metrics, and meter_provider into Client - Instrument Connection.api_request for per-attempt HTTP duration, TTFB, retries, and errors - Add _start_storage_operation, _PropertyMixin, and lazy HTTPIterator page metrics helpers - Skip per-attempt HTTP metrics while queuing sub-requests into Batch - Suppress internal background BucketMetadataCache lookups from emitting user-facing metrics
… and resource APIs - Instrument logical operations on Client and Bucket metadata, IAM, lifecycle, and list APIs - Record Batch.finish() logical operation metrics while deferring queued sub-requests - Instrument BucketACL, DefaultObjectACL, ObjectACL, HMACKeyMetadata, and BucketNotification
…ata APIs - Instrument Blob JSON metadata operations including reload, update, patch, delete, copy, rewrite, compose, and restore - Add logical operation metrics for Blob IAM policy and temporary/event-based hold helpers
…sfers - Instrument _media/requests downloads, simple/multipart/resumable uploads, and XML MPU operations - Record streaming download TTFB on first body chunk and logical payload body sizes once per operation - Instrument Blob download_to_*, download_as_*, and upload_from_* entrypoints - Preserve primary failure attribution when cleanup calls fail without mutating exception chains
….open streams - Add stream-lifetime logical operation metrics for BlobReader and BlobWriter - Propagate parent operation context across chunked reads, resumable uploads, and close flushes - Wrap Blob.open text streams with _MetricsTextIOWrapper to preserve stream-lifetime attribution
- Instrument upload_many, download_many, upload_many_from_filenames, and download_many_to_path - Instrument download_chunks_concurrently and upload_chunks_concurrently across THREAD and PROCESS workers - Propagate parent operation context across worker threads and record logical payload sizes
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces comprehensive OpenTelemetry metrics instrumentation across the Google Cloud Storage client, including tracking for latency, payload sizes, attempts, retries, and errors. Key feedback highlights potential ValueError crashes on detached buffers within _MetricsTextIOWrapper (suggesting a _safe_buffer helper) and a critical multiprocessing issue in transfer_manager where child processes could prematurely close parent connections during atexit cleanup.
| def _capture_text_op(self, fn, *args, **kwargs): | ||
| try: | ||
| result = fn(*args, **kwargs) | ||
| except BaseException as exc: | ||
| raw = self.buffer | ||
| if raw is not None: | ||
| raw._stream_error = exc | ||
| raise | ||
| self.buffer._stream_error = None | ||
| return result |
There was a problem hiding this comment.
Accessing self.buffer on a detached TextIOWrapper raises a ValueError: detached buffer. If an exception occurs inside _capture_text_op, accessing self.buffer in the except block will raise a ValueError, masking the original exception and making debugging extremely difficult.
To prevent this, we can define a _safe_buffer property that catches ValueError and returns None if the buffer is detached, and use it instead of self.buffer.
@property
def _safe_buffer(self):
try:
return self.buffer
except ValueError:
return None
def _capture_text_op(self, fn, *args, **kwargs):
try:
result = fn(*args, **kwargs)
except BaseException as exc:
raw = self._safe_buffer
if raw is not None:
raw._stream_error = exc
raise
raw = self._safe_buffer
if raw is not None:
raw._stream_error = None
return result| def __exit__(self, exc_type, exc_val, exc_tb): | ||
| raw = self.buffer | ||
| if exc_type is not None and isinstance(raw, BlobWriter) and not raw.closed: | ||
| raw.terminate() | ||
| return | ||
| super().__exit__(exc_type, exc_val, exc_tb) | ||
|
|
||
| def __del__(self): | ||
| raw = self.buffer | ||
| if raw is not None and not raw.closed: | ||
| raw._metrics_finalized = True | ||
| raw._buffer.close() |
There was a problem hiding this comment.
Similarly, accessing self.buffer inside __exit__ and __del__ can raise a ValueError: detached buffer if the stream has been detached. This can cause unexpected crashes during context exit or garbage collection. We should use the _safe_buffer property here as well.
| def __exit__(self, exc_type, exc_val, exc_tb): | |
| raw = self.buffer | |
| if exc_type is not None and isinstance(raw, BlobWriter) and not raw.closed: | |
| raw.terminate() | |
| return | |
| super().__exit__(exc_type, exc_val, exc_tb) | |
| def __del__(self): | |
| raw = self.buffer | |
| if raw is not None and not raw.closed: | |
| raw._metrics_finalized = True | |
| raw._buffer.close() | |
| def __exit__(self, exc_type, exc_val, exc_tb): | |
| raw = self._safe_buffer | |
| if exc_type is not None and isinstance(raw, BlobWriter) and not raw.closed: | |
| raw.terminate() | |
| return | |
| super().__exit__(exc_type, exc_val, exc_tb) | |
| def __del__(self): | |
| raw = self._safe_buffer | |
| if raw is not None and not raw.closed: | |
| raw._metrics_finalized = True | |
| raw._buffer.close() |
| def _cleanup_cached_clients(): | ||
| if not _cached_clients: | ||
| return | ||
| clients = list(_cached_clients.values()) | ||
| _cached_clients.clear() | ||
| had_active_metrics = False | ||
| for cl in clients: | ||
| if getattr(cl, "metrics_enabled", False) or getattr( | ||
| cl, "debug_metrics_enabled", False | ||
| ): | ||
| had_active_metrics = True | ||
| try: | ||
| cl.close() | ||
| except Exception: | ||
| pass | ||
| if not had_active_metrics: | ||
| return | ||
| metrics_mod = _opentelemetry_metrics.metrics | ||
| if metrics_mod is not None: | ||
| try: | ||
| provider = metrics_mod.get_meter_provider() | ||
| if hasattr(provider, "force_flush"): | ||
| provider.force_flush() | ||
| except Exception: | ||
| pass |
There was a problem hiding this comment.
In a multiprocess environment, when a child process is forked, it inherits the registered atexit handlers and the _cached_clients dictionary. Upon exit, the child process will run _cleanup_cached_clients and close the parent's cached clients/connections. Since the underlying sockets are shared, this can disrupt active connections in the parent process.
To prevent this, we should check os.getpid() == _cleanup_registered_pid inside _cleanup_cached_clients to ensure we only clean up if we are in the process that registered the cleanup. Additionally, when closing multiple resources during cleanup, we should wrap each close call in a try-except block, log any failures as warnings, and use a finally block to clear the resource references. Avoid broad except blocks that silently ignore exceptions.
def _cleanup_cached_clients():
if os.getpid() != _cleanup_registered_pid:
return
if not _cached_clients:
return
clients = list(_cached_clients.values())
had_active_metrics = False
for cl in clients:
if getattr(cl, "metrics_enabled", False) or getattr(
cl, "debug_metrics_enabled", False
):
had_active_metrics = True
try:
for cl in clients:
try:
cl.close()
except Exception as exc:
import logging
logging.getLogger(__name__).warning("Failed to close client: %s", exc)
finally:
_cached_clients.clear()
if not had_active_metrics:
return
metrics_mod = _opentelemetry_metrics.metrics
if metrics_mod is not None:
try:
provider = metrics_mod.get_meter_provider()
if hasattr(provider, "force_flush"):
provider.force_flush()
except Exception as exc:
import logging
logging.getLogger(__name__).warning("Failed to flush metrics: %s", exc)References
- When closing multiple resources during cleanup, wrap each close call in a try-except block to handle exceptions individually, log any failures as warnings, and use a finally block to clear the resource references.
- Avoid broad except Exception: blocks that silently return None. Instead, log the exception to aid in debugging.
Summary
Adds OpenTelemetry standard metrics support across
transfer_managerbatch and concurrent chunk transfer APIs ingoogle-cloud-storage.Key Changes
transfer_manager.py): Instrumentsupload_many,download_many,upload_many_from_filenames, anddownload_many_to_pathacross bothTHREADandPROCESSworker pools, including_LazyClientmetrics propagation and shutdown flushing.transfer_manager.py): Instrumentsdownload_chunks_concurrently(ReadObject) andupload_chunks_concurrently(WriteObject), propagating parent operation context across worker threads and recording logical payload sizes.