Repository navigation
feat(storage): add OpenTelemetry standard metrics for HTTP media, fileio, and transfer_manager - #18899
feat(storage): add OpenTelemetry standard metrics for HTTP media, fileio, and transfer_manager#18899shradhakatyal wants to merge 8 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]
…eneer APIs - Wire _MetricsRecorder lifecycle and meter_provider configuration into Client - Instrument _StorageConnection.api_request for per-attempt HTTP duration, TTFB, retries, and errors - Add logical operation instrumentation across Client, Bucket, Blob, ACL, Batch, HMACKeyMetadata, and BucketNotification - Defer Batch sub-request operations until Batch.finish() and track lazy HTTPIterator page metrics - Suppress internal background bucket metadata cache lookups from emitting user-facing metrics
…eio, and transfer_manager - 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 - Add stream-lifetime metrics for BlobReader, BlobWriter, and Blob.open binary and text streams - Instrument transfer_manager batch and concurrent chunk helpers across THREAD and PROCESS workers - Preserve primary failure attribution when cleanup calls fail without mutating exception chains
![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 implements OpenTelemetry standard metrics instrumentation across the Google Cloud Storage Python client library, adding comprehensive tracking for operations, attempts, retries, latencies, and payload sizes. It introduces a dedicated metrics management module, instruments core components like the HTTP connection, media download/upload helpers, file I/O streams, and the concurrent transfer manager, and adds extensive unit tests. The review feedback highlights potential ValueError: detached buffer issues in _MetricsTextIOWrapper within fileio.py when accessing self.buffer after a stream has been detached, providing actionable code suggestions to handle these cases gracefully in _capture_text_op, __exit__, and __del__.
| 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.
In _capture_text_op, accessing self.buffer can raise a ValueError: detached buffer if the underlying stream has been detached. This would override the original exception exc with a ValueError or raise an unexpected ValueError on success.
We can make this robust by retrieving the buffer reference once at the start of the method and handling the potential ValueError gracefully.
| 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 | |
| def _capture_text_op(self, fn, *args, **kwargs): | |
| try: | |
| raw = self.buffer | |
| except ValueError: | |
| raw = None | |
| try: | |
| result = fn(*args, **kwargs) | |
| except BaseException as exc: | |
| if raw is not None: | |
| raw._stream_error = exc | |
| raise | |
| 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.
In __exit__ and __del__, accessing self.buffer can raise a ValueError: detached buffer if the stream has been detached. In __del__, this will print an ignored exception warning to sys.stderr, and in __exit__ it will prevent super().__exit__ from running.
We should wrap the self.buffer access in a try...except ValueError: block to handle detached buffers gracefully.
| 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): | |
| try: | |
| raw = self.buffer | |
| except ValueError: | |
| raw = None | |
| 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): | |
| try: | |
| raw = self.buffer | |
| except ValueError: | |
| raw = None | |
| if raw is not None and not raw.closed: | |
| raw._metrics_finalized = True | |
| raw._buffer.close() |
Summary
Adds experimental OpenTelemetry standard metrics support across HTTP media downloads/uploads (
_media/requests/),fileio(BlobReader,BlobWriter,Blob.open), andtransfer_manageringoogle-cloud-storage.Key Changes
_media/requests/_request_helpers.py,download.py,upload.py,blob.py): InstrumentsDownload,RawDownload,ChunkedDownload,RawChunkedDownload,SimpleUpload,MultipartUpload,ResumableUpload,XMLMPUContainer, andXMLMPUPartfor per-attempt HTTP duration, streaming first-bytegcp.storage.client.ttfb, retries, errors, and logical payload body sizes.fileio.py,blob.py): Records one logicalReadObjectorWriteObjectoperation across the full stream lifetime ofBlobReader,BlobWriter, andBlob.open()(binary and text modes), while attributing per-chunk HTTP attempts to the parent operation.transfer_manager.py): Instrumentsupload_many,download_many,upload_many_from_filenames,download_many_to_path,download_chunks_concurrently, andupload_chunks_concurrentlyacross bothTHREADandPROCESSworker pools, including_LazyClientmetrics propagation and shutdown flushing._helpers.py,blob.py,transfer_manager.py): Preserves primary upload/corruption failure attribution on logical metrics when cleanup calls fail without mutating caller exception chains.