Repository navigation
feat(storage): add OpenTelemetry standard metrics recorder, views, and helpers - #18753
shradhakatyal wants to merge 1 commit 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]
![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 for the Google Cloud Storage client, including custom histogram buckets, canonical error classification, URL template normalization, logical operation context tracking, and a metrics recorder class, along with extensive unit tests. The review feedback suggests several robustness improvements to the error classification logic, such as validating that exception args are iterable, handling mock objects gracefully during MRO and gRPC status name resolution, performing more precise module prefix matching, and classifying concurrent.futures.CancelledError as a cancellation.
| for arg in getattr(current, "args", ()): | ||
| if isinstance(arg, BaseException) and id(arg) not in visited: | ||
| visited.add(id(arg)) | ||
| queue.append(arg) |
There was a problem hiding this comment.
To prevent potential TypeError exceptions when traversing the exception chain (for example, if a mock exception or a custom exception with a non-tuple args attribute is passed), it is safer to verify that args is a tuple or list before iterating over it.
| for arg in getattr(current, "args", ()): | |
| if isinstance(arg, BaseException) and id(arg) not in visited: | |
| visited.add(id(arg)) | |
| queue.append(arg) | |
| args = getattr(current, "args", ()) | |
| if isinstance(args, (tuple, list)): | |
| for arg in args: | |
| if isinstance(arg, BaseException) and id(arg) not in visited: | |
| visited.add(id(arg)) | |
| queue.append(arg) |
| for base in getattr(cls, "__mro__", ()): | ||
| mod = getattr(base, "__module__", "") or "" | ||
| if mod.startswith(module_prefix): | ||
| if class_names is None or base.__name__ in class_names: | ||
| return True | ||
| return False |
There was a problem hiding this comment.
When checking class inheritance by module, if cls is a mock object (commonly used in tests), getattr(cls, "__mro__", ()) can return a mock object instead of a tuple, which will raise a TypeError when iterated. Additionally, using mod.startswith(module_prefix) can incorrectly match sub-modules or modules with a common prefix (e.g., matching google.cloud.storage_transfer when prefix is google.cloud.storage). We should check isinstance(mro, tuple) and perform a more precise module prefix check.
| for base in getattr(cls, "__mro__", ()): | |
| mod = getattr(base, "__module__", "") or "" | |
| if mod.startswith(module_prefix): | |
| if class_names is None or base.__name__ in class_names: | |
| return True | |
| return False | |
| mro = getattr(cls, "__mro__", None) | |
| if isinstance(mro, tuple): | |
| for base in mro: | |
| mod = getattr(base, "__module__", "") or "" | |
| if mod == module_prefix or mod.startswith(module_prefix + "."): | |
| if class_names is None or getattr(base, "__name__", None) in class_names: | |
| return True | |
| return False |
| for item in chain: | ||
| if isinstance(item, asyncio.CancelledError): | ||
| return "CANCELLED" |
There was a problem hiding this comment.
In addition to asyncio.CancelledError, we should also check for concurrent.futures.CancelledError to ensure that cancellations occurring in thread pools or background workers (such as those in transfer_manager) are correctly classified as "CANCELLED".
| for item in chain: | |
| if isinstance(item, asyncio.CancelledError): | |
| return "CANCELLED" | |
| for item in chain: | |
| if isinstance(item, (asyncio.CancelledError, concurrent.futures.CancelledError)): | |
| return "CANCELLED" |
| if rpc_status is not None: | ||
| name = getattr(rpc_status, "name", None) | ||
| if name: | ||
| return str(name) |
There was a problem hiding this comment.
When resolving the gRPC status code name, if rpc_status is a mock object, getattr(rpc_status, "name", None) will return a mock object which is truthy, leading to returning a string representation of the mock as the error type. We should explicitly verify that name is a string to prevent this.
| if rpc_status is not None: | |
| name = getattr(rpc_status, "name", None) | |
| if name: | |
| return str(name) | |
| if rpc_status is not None: | |
| name = getattr(rpc_status, "name", None) | |
| if isinstance(name, str): | |
| return name |
This branch has not been deployed
1 participant
Summary
Implements the core OpenTelemetry standard metrics infrastructure for Google Cloud Storage (
_opentelemetry_metrics.py), laying the foundation for gRPC and HTTP standard metrics instrumentation:_MetricsRecorder: Manages the 9 standard metric instruments (rpc.client.call.duration,http.client.request.duration,gcp.client.request.duration,gcp.storage.client.operations,gcp.storage.client.attempts,gcp.storage.client.request.body.size,gcp.storage.client.response.body.size,gcp.storage.client.operation.ttfb, andgcp.storage.client.errors), per-instrument attribute scoping, and customMeterProviderflushing onshutdown().get_histogram_views): Defines custom bucket boundaries for latency (165 buckets from0.0sto<300.0s) and byte size (45 buckets from0 Bto16 GiB).classify_error_type): Maps transport, authentication, checksum, timeout, SSL/TLS, DNS, socket, gRPC, and HTTP errors to low-cardinalityerror.typevalues.normalize_url_template): Normalizes GCS JSON, Media, Batch, and XML Multipart Upload URLs into low-cardinalityurl.templatestrings, falling back to"UNKNOWN".start_operation): Tracks the outermost user-invoked Veneer method viaContextVarto prevent duplicate logical operation metrics on nested internal calls and supportssuppress=Truefor internal chunk/prefetch sub-operations.tests/unit/test__opentelemetry_metrics.pywith 100% statement coverage.[Generated-by: AI]