Repository navigation
feat(storage): add OpenTelemetry standard metrics for fileio and Blob.open streams - #18944
Draft
shradhakatyal wants to merge 11 commits into
Draft
shradhakatyal wants to merge 11 commits into
shradhakatyal wants to merge 11 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
This was referenced Nov 1, 2026
Draft
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request integrates OpenTelemetry metrics instrumentation across various Google Cloud Storage (GCS) client operations, tracking logical operations, attempts, retries, errors, payload sizes, and TTFB. It also adds comprehensive unit tests to verify the metrics. The review feedback correctly identifies potential AttributeError bugs in _MetricsTextIOWrapper when accessing self.buffer after it has been detached, providing clear suggestions to check for None before setting _stream_error.
Comment on lines
+701
to
+710
| 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 |
Contributor
There was a problem hiding this comment.
In _capture_text_op, self.buffer is accessed directly as self.buffer._stream_error = None at the end of the method. However, self.buffer can be None if the wrapper has been detached (e.g., via detach()). To prevent an AttributeError, please check if self.buffer is not None before setting _stream_error.
Suggested change
| 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: | |
| result = fn(*args, **kwargs) | |
| except BaseException as exc: | |
| raw = self.buffer | |
| if raw is not None: | |
| raw._stream_error = exc | |
| raise | |
| raw = self.buffer | |
| if raw is not None: | |
| raw._stream_error = None | |
| return result |
Comment on lines
+718
to
+730
| def __next__(self): | ||
| try: | ||
| line = super().__next__() | ||
| except StopIteration: | ||
| self.buffer._stream_error = None | ||
| raise | ||
| except BaseException as exc: | ||
| raw = self.buffer | ||
| if raw is not None: | ||
| raw._stream_error = exc | ||
| raise | ||
| self.buffer._stream_error = None | ||
| return line |
Contributor
There was a problem hiding this comment.
Similarly, in __next__, self.buffer is accessed directly as self.buffer._stream_error = None in both the StopIteration block and at the end of the method. If the stream is detached, this will raise an AttributeError. Please check if self.buffer is not None before setting _stream_error.
Suggested change
| def __next__(self): | |
| try: | |
| line = super().__next__() | |
| except StopIteration: | |
| self.buffer._stream_error = None | |
| raise | |
| except BaseException as exc: | |
| raw = self.buffer | |
| if raw is not None: | |
| raw._stream_error = exc | |
| raise | |
| self.buffer._stream_error = None | |
| return line | |
| def __next__(self): | |
| try: | |
| line = super().__next__() | |
| except StopIteration: | |
| raw = self.buffer | |
| if raw is not None: | |
| raw._stream_error = None | |
| raise | |
| except BaseException as exc: | |
| raw = self.buffer | |
| if raw is not None: | |
| raw._stream_error = exc | |
| raise | |
| raw = self.buffer | |
| if raw is not None: | |
| raw._stream_error = None | |
| return line |
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Adds stream-lifetime OpenTelemetry standard metrics support for
BlobReader,BlobWriter, andBlob.open()binary and text streams ingoogle-cloud-storage.Key Changes
fileio.py,blob.py): Records one logicalReadObjectorWriteObjectoperation spanning open-through-close forBlobReader,BlobWriter, andBlob.open()._helpers.py,fileio.py): Adds_bind_suboperation_contextso per-chunk range reads, resumable upload chunks, andclose()flushes attribute attempts and transferred bytes to the parent stream operation.fileio.py,blob.py): WrapsBlob.open()text-mode streams with_MetricsTextIOWrapperso closing or encountering errors at the text wrapper layer finalizes the stream metric.