From 43d1e94caf36dce78162e9fe38a7c6c2fd1dd47f Mon Sep 17 00:00:00 2001 From: Shradha Katyal Date: Thu, 15 Oct 2026 15:44:24 +0000 Subject: [PATCH 1/3] feat(storage): add OpenTelemetry standard metrics recorder, views, and 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] --- .../cloud/storage/_opentelemetry_metrics.py | 959 ++++++++++++++++- .../tests/unit/test__opentelemetry_metrics.py | 981 +++++++++++++++++- 2 files changed, 1936 insertions(+), 4 deletions(-) diff --git a/packages/google-cloud-storage/google/cloud/storage/_opentelemetry_metrics.py b/packages/google-cloud-storage/google/cloud/storage/_opentelemetry_metrics.py index 8592cc24c65d..c357a968407f 100644 --- a/packages/google-cloud-storage/google/cloud/storage/_opentelemetry_metrics.py +++ b/packages/google-cloud-storage/google/cloud/storage/_opentelemetry_metrics.py @@ -14,9 +14,21 @@ """Manages OpenTelemetry metrics instruments and gating for GCS client.""" +import asyncio +from collections import deque +import concurrent.futures +import contextlib +from contextvars import ContextVar +from dataclasses import dataclass +import http.client import logging import os -from typing import Any, Dict, Optional +import re +import socket +import ssl +import time +from typing import Any, Deque, Dict, Iterator, List, Optional, Set, Tuple +from urllib.parse import urlsplit from google.cloud.storage.version import __version__ @@ -38,8 +50,35 @@ _DEFAULT_ENABLE_METRICS = False _DEFAULT_ENABLE_DEBUG_METRICS = False +DEFAULT_SERVER_ADDRESS = "storage.googleapis.com" + +# --------------------------------------------------------------------------- +# 3. Standard Metric Instrument Names # --------------------------------------------------------------------------- -# 3. Optional OpenTelemetry Dependency Check +METRIC_RPC_CLIENT_CALL_DURATION = "rpc.client.call.duration" +METRIC_HTTP_CLIENT_REQUEST_DURATION = "http.client.request.duration" +METRIC_GCP_CLIENT_REQUEST_DURATION = "gcp.client.request.duration" +METRIC_STORAGE_CLIENT_OPERATIONS = "gcp.storage.client.operations" +METRIC_STORAGE_CLIENT_ATTEMPTS = "gcp.storage.client.attempts" +METRIC_STORAGE_REQUEST_BODY_SIZE = "gcp.storage.client.request.body.size" +METRIC_STORAGE_RESPONSE_BODY_SIZE = "gcp.storage.client.response.body.size" +METRIC_STORAGE_OPERATION_TTFB = "gcp.storage.client.operation.ttfb" +METRIC_STORAGE_CLIENT_ERRORS = "gcp.storage.client.errors" + +_LATENCY_METRIC_NAMES = ( + METRIC_RPC_CLIENT_CALL_DURATION, + METRIC_HTTP_CLIENT_REQUEST_DURATION, + METRIC_GCP_CLIENT_REQUEST_DURATION, + METRIC_STORAGE_OPERATION_TTFB, +) + +_SIZE_METRIC_NAMES = ( + METRIC_STORAGE_REQUEST_BODY_SIZE, + METRIC_STORAGE_RESPONSE_BODY_SIZE, +) + +# --------------------------------------------------------------------------- +# 4. Optional OpenTelemetry Dependency Check # --------------------------------------------------------------------------- try: from opentelemetry import metrics @@ -134,7 +173,7 @@ def is_debug_metrics_enabled( # --------------------------------------------------------------------------- -# 4. Standard Common Attributes & Meter Provider +# 5. Standard Common Attributes & Meter Provider # --------------------------------------------------------------------------- _COMMON_ATTRIBUTES: Dict[str, Any] = { "gcp.client.service": "storage", @@ -178,3 +217,917 @@ def get_meter( __version__, meter_provider=meter_provider, ) + + +# --------------------------------------------------------------------------- +# 6. Custom Histogram Bucket Boundaries & Views +# --------------------------------------------------------------------------- +def _build_latency_buckets() -> Tuple[float, ...]: + """Builds the custom latency histogram bucket boundaries in seconds. + + Uses 2ms resolution for the first 50 buckets (0.000s to 0.098s), then starts + at 0.100s with 10ms resolution and doubles the increment every 10 buckets + while boundary < 300.0s. + """ + boundaries: List[float] = [] + boundary = 0.0 + increment = 0.002 + + for _ in range(50): + boundaries.append(round(boundary, 6)) + boundary += increment + + increment = 0.010 + for i in range(150): + if boundary >= 300.0: + break + if i != 0 and i % 10 == 0: + increment *= 2 + boundaries.append(round(boundary, 6)) + boundary += increment + + return tuple(boundaries) + + +def _build_size_buckets() -> Tuple[int, ...]: + """Builds the custom byte-size histogram bucket boundaries in bytes. + + Uses 128 KiB resolution from 0 up to 4 MiB (33 buckets), then doubles every + bucket up to 16 GiB (12 buckets, 45 total buckets). + """ + kb = 1024 + mb = 1024 * kb + gb = 1024 * mb + + boundaries: List[int] = [] + boundary = 0 + increment = 128 * kb + + while boundary <= 16 * gb: + boundaries.append(boundary) + if boundary < 4 * mb: + boundary += increment + else: + boundary *= 2 + + return tuple(boundaries) + + +LATENCY_HISTOGRAM_BOUNDARIES: Tuple[float, ...] = _build_latency_buckets() +SIZE_HISTOGRAM_BOUNDARIES: Tuple[int, ...] = _build_size_buckets() + + +def get_histogram_views() -> List[Any]: + """Returns OpenTelemetry SDK Views for custom latency and size histograms. + + If ``opentelemetry-sdk`` is not installed, returns an empty list. + """ + try: + from opentelemetry.sdk.metrics.view import ( + ExplicitBucketHistogramAggregation, + View, + ) + except ImportError: + return [] + + views: List[Any] = [] + for name in _LATENCY_METRIC_NAMES: + views.append( + View( + instrument_name=name, + aggregation=ExplicitBucketHistogramAggregation( + boundaries=LATENCY_HISTOGRAM_BOUNDARIES + ), + ) + ) + for name in _SIZE_METRIC_NAMES: + views.append( + View( + instrument_name=name, + aggregation=ExplicitBucketHistogramAggregation( + boundaries=SIZE_HISTOGRAM_BOUNDARIES + ), + ) + ) + return views + + +# --------------------------------------------------------------------------- +# 7. Canonical Error Classification (`error.type`) +# --------------------------------------------------------------------------- +_HTTP_STATUS_TO_ERROR_TYPE: Dict[int, str] = { + 400: "INVALID_ARGUMENT", + 401: "UNAUTHENTICATED", + 403: "PERMISSION_DENIED", + 404: "NOT_FOUND", + 408: "DEADLINE_EXCEEDED", + 409: "ABORTED", + 412: "FAILED_PRECONDITION", + 416: "OUT_OF_RANGE", + 429: "RESOURCE_EXHAUSTED", + 499: "CANCELLED", + 500: "INTERNAL", + 501: "UNIMPLEMENTED", + 503: "UNAVAILABLE", + 504: "DEADLINE_EXCEEDED", +} + + +def _is_valid_status_code(val: Any) -> bool: + """Returns True if ``val`` is a valid HTTP error status code (int in 400..599).""" + return isinstance(val, int) and not isinstance(val, bool) and 400 <= val < 600 + + +def _iter_exception_chain(exc: BaseException) -> Iterator[BaseException]: + """Yields ``exc`` and any nested causes, contexts, reasons, or exception args.""" + visited: Set[int] = {id(exc)} + queue: Deque[BaseException] = deque([exc]) + while queue: + current = queue.popleft() + yield current + + for attr_name in ( + "__cause__", + "__context__", + "cause", + "reason", + "original_exception", + ): + nested = getattr(current, attr_name, None) + if isinstance(nested, BaseException) and id(nested) not in visited: + visited.add(id(nested)) + queue.append(nested) + + for arg in getattr(current, "args", ()): + if isinstance(arg, BaseException) and id(arg) not in visited: + visited.add(id(arg)) + queue.append(arg) + + +def _is_subclass_by_module( + cls: type, module_prefix: str, class_names: Optional[Set[str]] = None +) -> bool: + """Checks whether ``cls`` or any of its base classes match module/name criteria.""" + 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 + + +def classify_error_type( + exc: Optional[BaseException] = None, + status_code: Optional[int] = None, +) -> str: + """Maps an exception or HTTP status code to a low-cardinality ``error.type``. + + Args: + exc: Optional exception raised during an operation or transport attempt. + status_code: Optional HTTP status code from the response. + + Returns: + str: Canonical low-cardinality error classification string. + """ + valid_status = _is_valid_status_code(status_code) + if exc is None: + if not valid_status or status_code is None: + return "OK" + return _HTTP_STATUS_TO_ERROR_TYPE.get(status_code, str(status_code)) + + chain = list(_iter_exception_chain(exc)) + + # 1. Checksum Mismatch (DataCorruption) + for item in chain: + if _is_subclass_by_module( + item.__class__, "google.cloud.storage", {"DataCorruption"} + ) or _is_subclass_by_module( + item.__class__, "google.resumable_media", {"DataCorruption"} + ): + return "CHECKSUM_MISMATCH" + + # 2. Authentication / Credentials Failure (google.auth.exceptions) + for item in chain: + if _is_subclass_by_module( + item.__class__, + "google.auth.exceptions", + { + "RefreshError", + "TransportError", + "DefaultCredentialsError", + "GoogleAuthError", + }, + ): + return "AUTHENTICATION_ERROR" + + # 3. Cancellation + for item in chain: + if isinstance(item, asyncio.CancelledError): + return "CANCELLED" + + # 4. HTTP response status code on exception (checked before grpc_status_code so + # HTTP 409 Conflict consistently maps to ABORTED across HTTP and Veneer metrics) + for item in chain: + resp = getattr(item, "response", None) + resp_status = getattr(resp, "status_code", None) + if _is_valid_status_code(resp_status): + return _HTTP_STATUS_TO_ERROR_TYPE.get(resp_status, str(resp_status)) + + if item.__class__.__name__ == "Conflict" and _is_subclass_by_module( + item.__class__, "google.api_core.exceptions", {"Conflict"} + ): + return _HTTP_STATUS_TO_ERROR_TYPE[409] + + # 5. gRPC / GoogleAPICallError status code + for item in chain: + grpc_code = getattr(item, "grpc_status_code", None) + if grpc_code is not None: + name = getattr(grpc_code, "name", None) + if name: + return str(name) + return str(grpc_code) + + # Direct grpc.RpcError (has callable .code()) + code_attr = getattr(item, "code", None) + if callable(code_attr): + try: + rpc_status = code_attr() + if rpc_status is not None: + name = getattr(rpc_status, "name", None) + if name: + return str(name) + except Exception: # pragma: no cover + pass + + if _is_valid_status_code(code_attr): + return _HTTP_STATUS_TO_ERROR_TYPE.get(code_attr, str(code_attr)) + + # 6. TLS / SSL Failure + for item in chain: + if ( + isinstance(item, ssl.SSLError) + or _is_subclass_by_module( + item.__class__, + "requests.exceptions", + {"SSLError"}, + ) + or _is_subclass_by_module( + item.__class__, + "urllib3.exceptions", + {"SSLError"}, + ) + ): + return "TLS_FAILURE" + + # 7. Specific Socket / Network Failures (DNS, Connection Refused, Connection Reset) + for item in chain: + if isinstance(item, socket.gaierror) or _is_subclass_by_module( + item.__class__, + "urllib3.exceptions", + {"NameResolutionError"}, + ): + return "DNS_FAILURE" + if isinstance(item, ConnectionRefusedError): + return "CONNECTION_REFUSED" + if isinstance(item, ConnectionResetError): + return "CONNECTION_RESET" + + # 8. Timeouts (Note: urllib3.exceptions.NewConnectionError subclasses ConnectTimeoutError + # even for non-timeout connection failures, so exclude NewConnectionError here.) + for item in chain: + if item.__class__.__name__ != "NewConnectionError": + if isinstance( + item, + ( + socket.timeout, + TimeoutError, + asyncio.TimeoutError, + concurrent.futures.TimeoutError, + ), + ): + return "TIMEOUT" + if _is_subclass_by_module( + item.__class__, + "requests.exceptions", + {"Timeout", "ConnectTimeout", "ReadTimeout"}, + ) or _is_subclass_by_module( + item.__class__, + "urllib3.exceptions", + {"TimeoutError", "ReadTimeoutError", "ConnectTimeoutError"}, + ): + return "TIMEOUT" + + # 9. Generic Connectivity Failure + for item in chain: + if ( + _is_subclass_by_module( + item.__class__, + "requests.exceptions", + {"ConnectionError", "ChunkedEncodingError", "RequestException"}, + ) + or _is_subclass_by_module( + item.__class__, + "urllib3.exceptions", + {"HTTPError"}, + ) + or isinstance(item, (OSError, http.client.HTTPException)) + ): + return "CONNECTIVITY" + + # 10. Explicit HTTP status_code fallback + if valid_status and status_code is not None: + return _HTTP_STATUS_TO_ERROR_TYPE.get(status_code, str(status_code)) + + # 11. Client Validation Errors + for item in chain: + if ( + isinstance(item, (ValueError, TypeError)) + or _is_subclass_by_module( + item.__class__, "google.cloud.storage", {"InvalidResponse"} + ) + or _is_subclass_by_module( + item.__class__, "google.resumable_media", {"InvalidResponse"} + ) + ): + return "CLIENT_VALIDATION_ERROR" + + return "UNKNOWN" + + +# --------------------------------------------------------------------------- +# 8. HTTP `url.template` Route Normalization +# --------------------------------------------------------------------------- +_URL_TEMPLATE_PATTERNS: Tuple[Tuple[re.Pattern[str], str], ...] = ( + ( + re.compile(r"^/storage/v1/b/[^/]+/o/[^/]+/rewriteTo/b/[^/]+/o/.+$"), + "/storage/v1/b/{bucket}/o/{object}/rewriteTo/b/{bucket}/o/{object}", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+/o/[^/]+/copyTo/b/[^/]+/o/.+$"), + "/storage/v1/b/{bucket}/o/{object}/copyTo/b/{bucket}/o/{object}", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+/o/[^/]+/moveTo/o/.+$"), + "/storage/v1/b/{bucket}/o/{object}/moveTo/o/{object}", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+/o/[^/]+/compose$"), + "/storage/v1/b/{bucket}/o/{object}/compose", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+/o/[^/]+/restore$"), + "/storage/v1/b/{bucket}/o/{object}/restore", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+/o/[^/]+/acl/[^/]+$"), + "/storage/v1/b/{bucket}/o/{object}/acl/{entity}", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+/o/[^/]+/acl$"), + "/storage/v1/b/{bucket}/o/{object}/acl", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+/o/[^/]+/iam(?:/.*)?$"), + "/storage/v1/b/{bucket}/o/{object}/iam", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+/o/.+$"), + "/storage/v1/b/{bucket}/o/{object}", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+/o$"), + "/storage/v1/b/{bucket}/o", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+/acl/[^/]+$"), + "/storage/v1/b/{bucket}/acl/{entity}", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+/acl$"), + "/storage/v1/b/{bucket}/acl", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+/defaultObjectAcl/[^/]+$"), + "/storage/v1/b/{bucket}/defaultObjectAcl/{entity}", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+/defaultObjectAcl$"), + "/storage/v1/b/{bucket}/defaultObjectAcl", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+/iam(?:/.*)?$"), + "/storage/v1/b/{bucket}/iam", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+/notificationConfigs/[^/]+$"), + "/storage/v1/b/{bucket}/notificationConfigs/{notification}", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+/notificationConfigs$"), + "/storage/v1/b/{bucket}/notificationConfigs", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+/lockRetentionPolicy$"), + "/storage/v1/b/{bucket}/lockRetentionPolicy", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+/restore$"), + "/storage/v1/b/{bucket}/restore", + ), + ( + re.compile(r"^/storage/v1/b/[^/]+$"), + "/storage/v1/b/{bucket}", + ), + ( + re.compile(r"^/storage/v1/b$"), + "/storage/v1/b", + ), + ( + re.compile(r"^/storage/v1/projects/[^/]+/serviceAccount$"), + "/storage/v1/projects/{project}/serviceAccount", + ), + ( + re.compile(r"^/storage/v1/projects/[^/]+/hmacKeys/[^/]+$"), + "/storage/v1/projects/{project}/hmacKeys/{access_id}", + ), + ( + re.compile(r"^/storage/v1/projects/[^/]+/hmacKeys$"), + "/storage/v1/projects/{project}/hmacKeys", + ), + ( + re.compile(r"^/upload/storage/v1/b/[^/]+/o(?:/[^/]+)?$"), + "/upload/storage/v1/b/{bucket}/o", + ), + ( + re.compile(r"^/download/storage/v1/b/[^/]+/o/.+$"), + "/download/storage/v1/b/{bucket}/o/{object}", + ), + ( + re.compile(r"^/batch/storage/v1$"), + "/batch/storage/v1", + ), +) + +_XML_MPU_RESERVED_PREFIXES = {"storage", "upload", "download", "batch", "b", "projects"} +_XML_MPU_QUERY_KEYS = {"uploads", "uploadId", "partNumber"} +_XML_MULTI_SEGMENT_RE = re.compile(r"^/([^/]+)/(.+)$") + + +def normalize_url_template(url_or_path: Optional[str]) -> str: + """Normalizes a raw HTTP URL or path into a bounded ``url.template`` string. + + Bucket names, object keys, project IDs, and query strings are stripped to + prevent cardinality explosion. Unrecognized paths fall back to ``"UNKNOWN"``. + + Args: + url_or_path: Full request URL or path string. + + Returns: + str: Low-cardinality ``url.template`` value. + """ + if not url_or_path or not isinstance(url_or_path, str): + return "UNKNOWN" + + parsed = urlsplit(url_or_path) + path = parsed.path or "" + query = parsed.query or "" + + path = "/" + path.strip("/") + if path == "/": + return "UNKNOWN" + + # Normalize relative paths passed to Connection.api_request before /storage/v1 prefixing + if path == "/b" or path.startswith(("/b/", "/projects/")): + path = "/storage/v1" + path + + for pattern, template in _URL_TEMPLATE_PATTERNS: + if pattern.match(path): + return template + + # XML API Multipart Upload routes: /{bucket}/{object} with XML MPU query params + if query: + query_keys = {param.partition("=")[0] for param in query.split("&") if param} + if query_keys & _XML_MPU_QUERY_KEYS: + match_multi = _XML_MULTI_SEGMENT_RE.match(path) + if match_multi and match_multi.group(1) not in _XML_MPU_RESERVED_PREFIXES: + return "/{bucket}/{object}" + + return "UNKNOWN" + + +# --------------------------------------------------------------------------- +# 9. Logical Operation Context & Suppression (`ContextVar`) +# --------------------------------------------------------------------------- +@dataclass(frozen=True) +class _OperationState: + """Represents the active logical GCS Veneer operation in the current context.""" + + client_method: str + rpc_method: str + suppressed: bool = False + + +_OP_STATE: ContextVar[Optional[_OperationState]] = ContextVar( + "_gcs_otel_op_state", default=None +) + + +def get_current_operation() -> Optional[_OperationState]: + """Returns the current logical `_OperationState` in this execution context, if any.""" + return _OP_STATE.get() + + +@contextlib.contextmanager +def start_operation( + recorder: Optional["_MetricsRecorder"], + client_method: str, + rpc_method: str, + suppress: bool = False, +) -> Iterator[Optional[_OperationState]]: + """Context manager tracking a logical user-facing GCS operation. + + If a parent logical operation is already active in the current context, + nested calls become a no-op so only the outermost user-invoked method + emits ``gcp.client.request.duration`` and ``gcp.storage.client.operations``. + + Args: + recorder: Optional `_MetricsRecorder` configured on the client. + client_method: Fully qualified Python method name + (e.g. ``"storage.Blob.download_to_filename"``). + rpc_method: Canonical GCS RPC operation name + (e.g. ``"storage.objects.get"``). + suppress: If True, suppresses logical operation and payload size + emission for this context and any nested calls (used by + ``transfer_manager`` workers and ``BlobReader`` prefetches). + """ + existing = _OP_STATE.get() + if existing is not None: + if suppress and not existing.suppressed: + suppressed_child = _OperationState( + client_method=existing.client_method, + rpc_method=existing.rpc_method, + suppressed=True, + ) + child_token = _OP_STATE.set(suppressed_child) + try: + yield suppressed_child + finally: + _OP_STATE.reset(child_token) + else: + yield existing + return + + recorder_active = recorder is not None and recorder.standard_enabled + if not recorder_active and not suppress: + yield None + return + + state = _OperationState( + client_method=client_method, + rpc_method=rpc_method, + suppressed=suppress, + ) + token = _OP_STATE.set(state) + start_time = time.perf_counter() + error_type = "OK" + try: + yield state + except BaseException as exc: + error_type = classify_error_type(exc) + raise + finally: + duration = max(0.0, time.perf_counter() - start_time) + _OP_STATE.reset(token) + if recorder_active and not suppress and recorder is not None: + recorder.record_logical_operation( + duration=duration, + client_method=client_method, + rpc_method=rpc_method, + error_type=error_type, + ) + + +# --------------------------------------------------------------------------- +# 10. Server Address Parsing & `_MetricsRecorder` +# --------------------------------------------------------------------------- +def _extract_server_address(api_endpoint: Optional[str]) -> str: + """Extracts the low-cardinality hostname from ``api_endpoint``.""" + if not api_endpoint or not isinstance(api_endpoint, str): + return DEFAULT_SERVER_ADDRESS + endpoint = api_endpoint.strip() + if not endpoint: + return DEFAULT_SERVER_ADDRESS + if "://" not in endpoint: + endpoint = "https://" + endpoint.lstrip("/") + parsed = urlsplit(endpoint) + return parsed.hostname or DEFAULT_SERVER_ADDRESS + + +class _MetricsRecorder: + """Holds OpenTelemetry metric instruments and records standard GCS metrics.""" + + def __init__( + self, + enable_metrics: Optional[bool] = None, + enable_debug_metrics: Optional[bool] = None, + meter_provider: Optional[Any] = None, + api_endpoint: Optional[str] = None, + rpc_system: str = "http", + ) -> None: + self.standard_enabled: bool = is_metrics_enabled(enable_metrics) + self.debug_enabled: bool = is_debug_metrics_enabled(enable_debug_metrics) + self.enabled: bool = self.standard_enabled or self.debug_enabled + self._meter_provider: Optional[Any] = meter_provider + self.rpc_system: str = rpc_system + self.server_address: str = _extract_server_address(api_endpoint) + self.meter: Optional[Any] = get_meter( + enable_metrics=enable_metrics, + enable_debug_metrics=enable_debug_metrics, + meter_provider=meter_provider, + ) + + self._base_attributes: Dict[str, Any] = { + **_COMMON_ATTRIBUTES, + "server.address": self.server_address, + "rpc.system.name": self.rpc_system, + } + + if self.standard_enabled and self.meter is not None: + self.rpc_call_duration = self.meter.create_histogram( + METRIC_RPC_CLIENT_CALL_DURATION, + unit="s", + description="Duration of an individual gRPC attempt in seconds (excluding retries).", + ) + self.http_request_duration = self.meter.create_histogram( + METRIC_HTTP_CLIENT_REQUEST_DURATION, + unit="s", + description="Duration of an individual HTTP request attempt in seconds (excluding retries).", + ) + self.client_request_duration = self.meter.create_histogram( + METRIC_GCP_CLIENT_REQUEST_DURATION, + unit="s", + description="End-to-end latency of a logical user-facing Veneer operation in seconds.", + ) + self.operations_counter = self.meter.create_counter( + METRIC_STORAGE_CLIENT_OPERATIONS, + unit="{operation}", + description="Number of logical GCS client operations.", + ) + self.attempts_counter = self.meter.create_counter( + METRIC_STORAGE_CLIENT_ATTEMPTS, + unit="{attempt}", + description="Number of physical network transport attempts executed.", + ) + self.request_body_size = self.meter.create_histogram( + METRIC_STORAGE_REQUEST_BODY_SIZE, + unit="By", + description="Application payload bytes submitted by the caller for an upload or write operation.", + ) + self.response_body_size = self.meter.create_histogram( + METRIC_STORAGE_RESPONSE_BODY_SIZE, + unit="By", + description="Application payload bytes delivered to the caller.", + ) + self.operation_ttfb = self.meter.create_histogram( + METRIC_STORAGE_OPERATION_TTFB, + unit="s", + description="Time elapsed from attempt initiation to receipt of the first response byte.", + ) + self.errors_counter = self.meter.create_counter( + METRIC_STORAGE_CLIENT_ERRORS, + unit="{error}", + description="Number of physical network transport attempts terminating with a non-OK status.", + ) + else: + self.rpc_call_duration = None + self.http_request_duration = None + self.client_request_duration = None + self.operations_counter = None + self.attempts_counter = None + self.request_body_size = None + self.response_body_size = None + self.operation_ttfb = None + self.errors_counter = None + + def get_base_attributes(self) -> Dict[str, Any]: + """Returns a copy of the base attributes for this recorder instance.""" + return self._base_attributes.copy() + + def _build_operation_attributes( + self, + operation: Optional[str] = None, + client_method: Optional[str] = None, + current_op: Optional[_OperationState] = None, + ) -> Dict[str, Any]: + """Builds base attributes enriched with current or explicit operation metadata.""" + attrs = self._base_attributes.copy() + if current_op is None: + current_op = _OP_STATE.get() + resolved_op = operation or (current_op.rpc_method if current_op else None) + resolved_method = client_method or ( + current_op.client_method if current_op else None + ) + if resolved_op is not None: + attrs["gcp.storage.operation"] = resolved_op + if resolved_method is not None: + attrs["gcp.client.method"] = resolved_method + return attrs + + def record_logical_operation( + self, + duration: float, + client_method: str, + rpc_method: str, + error_type: str = "OK", + ) -> None: + """Records end-to-end logical operation duration and count.""" + if not self.standard_enabled or self.meter is None: + return + attrs = { + **self._base_attributes, + "gcp.storage.operation": rpc_method, + "gcp.client.method": client_method, + "error.type": error_type, + } + self.client_request_duration.record(max(0.0, duration), attributes=attrs) + self.operations_counter.add(1, attributes=attrs) + + def record_http_attempt( + self, + duration: float, + http_method: str, + url_template: str, + status_code: Optional[int] = None, + error_type: str = "OK", + operation: Optional[str] = None, + client_method: Optional[str] = None, + network_path: Optional[str] = None, + ) -> None: + """Records a single physical HTTP request attempt.""" + if not self.standard_enabled or self.meter is None: + return + current_op = _OP_STATE.get() + resolved_op = operation or (current_op.rpc_method if current_op else None) + resolved_method = client_method or ( + current_op.client_method if current_op else None + ) + valid_code = ( + isinstance(status_code, int) + and not isinstance(status_code, bool) + and 100 <= status_code < 600 + ) + + # 1. http.client.request.duration attributes + http_attrs: Dict[str, Any] = { + **self._base_attributes, + "http.request.method": http_method.upper() if http_method else "UNKNOWN", + "url.template": url_template, + "error.type": error_type, + } + if valid_code: + http_attrs["http.response.status_code"] = status_code + if network_path is not None: + http_attrs["network.path"] = network_path + + # 2. gcp.storage.client.attempts attributes + attempt_attrs: Dict[str, Any] = { + **self._base_attributes, + "error.type": error_type, + } + if resolved_op is not None: + attempt_attrs["gcp.storage.operation"] = resolved_op + if valid_code: + attempt_attrs["http.response.status_code"] = status_code + if network_path is not None: + attempt_attrs["network.path"] = network_path + + self.http_request_duration.record(max(0.0, duration), attributes=http_attrs) + self.attempts_counter.add(1, attributes=attempt_attrs) + + # 3. gcp.storage.client.errors attributes + if error_type != "OK": + error_attrs: Dict[str, Any] = { + **self._base_attributes, + "error.type": error_type, + } + if resolved_op is not None: + error_attrs["gcp.storage.operation"] = resolved_op + if resolved_method is not None: + error_attrs["gcp.client.method"] = resolved_method + self.errors_counter.add(1, attributes=error_attrs) + + def record_rpc_attempt( + self, + duration: float, + operation: str, + status_code: Optional[str] = None, + error_type: str = "OK", + client_method: Optional[str] = None, + network_path: Optional[str] = None, + ) -> None: + """Records a single physical gRPC call attempt.""" + if not self.standard_enabled or self.meter is None: + return + op_attrs = self._build_operation_attributes( + operation=operation, client_method=client_method + ) + + # 1. rpc.client.call.duration attributes + rpc_attrs: Dict[str, Any] = { + **op_attrs, + "error.type": error_type, + } + if status_code is not None: + rpc_attrs["rpc.response.status_code"] = status_code + if network_path is not None: + rpc_attrs["network.path"] = network_path + + # 2. gcp.storage.client.attempts attributes + attempt_attrs: Dict[str, Any] = { + **self._base_attributes, + "gcp.storage.operation": operation, + "error.type": error_type, + } + if status_code is not None: + attempt_attrs["rpc.response.status_code"] = status_code + if network_path is not None: + attempt_attrs["network.path"] = network_path + + self.rpc_call_duration.record(max(0.0, duration), attributes=rpc_attrs) + self.attempts_counter.add(1, attributes=attempt_attrs) + + # 3. gcp.storage.client.errors attributes + if error_type != "OK": + error_attrs: Dict[str, Any] = { + **op_attrs, + "error.type": error_type, + } + self.errors_counter.add(1, attributes=error_attrs) + + def record_request_body_size( + self, + size: int, + operation: Optional[str] = None, + client_method: Optional[str] = None, + ) -> None: + """Records the payload byte size of an upload or write operation.""" + if not self.standard_enabled or self.meter is None: + return + if not isinstance(size, int) or isinstance(size, bool): + return + current_op = _OP_STATE.get() + if current_op is not None and current_op.suppressed: + return + attrs = self._build_operation_attributes( + operation=operation, client_method=client_method, current_op=current_op + ) + self.request_body_size.record(max(0, size), attributes=attrs) + + def record_response_body_size( + self, + size: int, + operation: Optional[str] = None, + client_method: Optional[str] = None, + ) -> None: + """Records the payload byte size of a download or read operation.""" + if not self.standard_enabled or self.meter is None: + return + if not isinstance(size, int) or isinstance(size, bool): + return + current_op = _OP_STATE.get() + if current_op is not None and current_op.suppressed: + return + attrs = self._build_operation_attributes( + operation=operation, client_method=client_method, current_op=current_op + ) + self.response_body_size.record(max(0, size), attributes=attrs) + + def record_ttfb( + self, + duration: float, + operation: Optional[str] = None, + client_method: Optional[str] = None, + ) -> None: + """Records time to first response byte (TTFB) in seconds.""" + if not self.standard_enabled or self.meter is None: + return + current_op = _OP_STATE.get() + if current_op is not None and current_op.suppressed: + return + attrs = self._build_operation_attributes( + operation=operation, client_method=client_method, current_op=current_op + ) + self.operation_ttfb.record(max(0.0, duration), attributes=attrs) + + def shutdown(self) -> None: + """Flushes pending metrics on a custom MeterProvider if configured.""" + if self._meter_provider is not None and hasattr( + self._meter_provider, "force_flush" + ): + try: + self._meter_provider.force_flush() + except Exception as exc: + logger.debug( + "Failed to force_flush custom MeterProvider during client shutdown: %s", + exc, + ) diff --git a/packages/google-cloud-storage/tests/unit/test__opentelemetry_metrics.py b/packages/google-cloud-storage/tests/unit/test__opentelemetry_metrics.py index 966cb8a1c78d..cb392b2f9870 100644 --- a/packages/google-cloud-storage/tests/unit/test__opentelemetry_metrics.py +++ b/packages/google-cloud-storage/tests/unit/test__opentelemetry_metrics.py @@ -12,13 +12,24 @@ # See the License for the specific language governing permissions and # limitations under the License. +import asyncio +import concurrent.futures +import http.client import importlib +import socket +import ssl import sys +import types from unittest import mock +from google.api_core import exceptions as api_exceptions +from google.auth import exceptions as auth_exceptions import pytest +import requests.exceptions +import urllib3.exceptions from google.cloud.storage import _opentelemetry_metrics +from google.cloud.storage import exceptions as storage_exceptions from google.cloud.storage.version import __version__ @@ -144,7 +155,6 @@ def test_client_setting_overrides_env_var(monkeypatch): def test_non_boolean_client_setting_raises_type_error(): """Passing a non-boolean setting such as 'false' must raise TypeError.""" - # Argument validation executes before the _ENABLE_METRICS_DEV_GATE check. with pytest.raises(TypeError, match="enable_metrics must be a boolean or None"): _opentelemetry_metrics.is_metrics_enabled(enable_metrics="false") # type: ignore[arg-type] @@ -287,3 +297,972 @@ def test_get_meter(monkeypatch): __version__, meter_provider=mock_provider, ) + + +# --------------------------------------------------------------------------- +# Histogram Bucket Boundaries & Views Tests +# --------------------------------------------------------------------------- +def test_histogram_bucket_boundaries(): + """Verify custom latency and byte-size bucket boundaries match specification.""" + latency = _opentelemetry_metrics.LATENCY_HISTOGRAM_BOUNDARIES + assert latency[0] == 0.0 + assert latency[1] == 0.002 + assert latency[49] == 0.098 + assert latency[50] == 0.1 + assert latency[60] == 0.2 + assert latency[-1] < 300.0 + assert list(latency) == sorted(latency) + + sizes = _opentelemetry_metrics.SIZE_HISTOGRAM_BOUNDARIES + assert len(sizes) == 45 + assert sizes[0] == 0 + assert sizes[1] == 128 * 1024 + assert sizes[32] == 4 * 1024 * 1024 + assert sizes[33] == 8 * 1024 * 1024 + assert sizes[-1] == 16 * 1024 * 1024 * 1024 + assert list(sizes) == sorted(sizes) + + +def test_get_histogram_views_with_and_without_sdk(monkeypatch): + """Verify get_histogram_views builds 6 Views when SDK is installed and [] otherwise.""" + fake_view_mod = types.ModuleType("opentelemetry.sdk.metrics.view") + + class FakeAggregation: + def __init__(self, boundaries): + self.boundaries = boundaries + + class FakeView: + def __init__(self, instrument_name, aggregation): + self.instrument_name = instrument_name + self.aggregation = aggregation + + fake_view_mod.ExplicitBucketHistogramAggregation = FakeAggregation + fake_view_mod.View = FakeView + + with monkeypatch.context() as m: + m.setitem(sys.modules, "opentelemetry", types.ModuleType("opentelemetry")) + m.setitem( + sys.modules, "opentelemetry.sdk", types.ModuleType("opentelemetry.sdk") + ) + m.setitem( + sys.modules, + "opentelemetry.sdk.metrics", + types.ModuleType("opentelemetry.sdk.metrics"), + ) + m.setitem(sys.modules, "opentelemetry.sdk.metrics.view", fake_view_mod) + + views = _opentelemetry_metrics.get_histogram_views() + assert len(views) == 6 + latency_views = [ + v + for v in views + if v.instrument_name in _opentelemetry_metrics._LATENCY_METRIC_NAMES + ] + size_views = [ + v + for v in views + if v.instrument_name in _opentelemetry_metrics._SIZE_METRIC_NAMES + ] + assert len(latency_views) == 4 + assert len(size_views) == 2 + assert ( + latency_views[0].aggregation.boundaries + == _opentelemetry_metrics.LATENCY_HISTOGRAM_BOUNDARIES + ) + assert ( + size_views[0].aggregation.boundaries + == _opentelemetry_metrics.SIZE_HISTOGRAM_BOUNDARIES + ) + + # Without opentelemetry-sdk installed + with monkeypatch.context() as m: + m.setitem(sys.modules, "opentelemetry.sdk.metrics.view", None) + assert _opentelemetry_metrics.get_histogram_views() == [] + + +# --------------------------------------------------------------------------- +# Error Classification (`classify_error_type`) Tests +# --------------------------------------------------------------------------- +def test_classify_error_type_none_and_status_codes(): + assert _opentelemetry_metrics.classify_error_type() == "OK" + assert _opentelemetry_metrics.classify_error_type(None, status_code=200) == "OK" + assert _opentelemetry_metrics.classify_error_type(None, status_code=206) == "OK" + assert ( + _opentelemetry_metrics.classify_error_type(None, status_code=True) == "OK" # type: ignore[arg-type] + ) + assert ( + _opentelemetry_metrics.classify_error_type(None, status_code=400) + == "INVALID_ARGUMENT" + ) + assert ( + _opentelemetry_metrics.classify_error_type(None, status_code=401) + == "UNAUTHENTICATED" + ) + assert ( + _opentelemetry_metrics.classify_error_type(None, status_code=403) + == "PERMISSION_DENIED" + ) + assert ( + _opentelemetry_metrics.classify_error_type(None, status_code=404) == "NOT_FOUND" + ) + assert ( + _opentelemetry_metrics.classify_error_type(None, status_code=408) + == "DEADLINE_EXCEEDED" + ) + assert ( + _opentelemetry_metrics.classify_error_type(None, status_code=409) == "ABORTED" + ) + assert ( + _opentelemetry_metrics.classify_error_type(None, status_code=412) + == "FAILED_PRECONDITION" + ) + assert ( + _opentelemetry_metrics.classify_error_type(None, status_code=416) + == "OUT_OF_RANGE" + ) + assert ( + _opentelemetry_metrics.classify_error_type(None, status_code=429) + == "RESOURCE_EXHAUSTED" + ) + assert ( + _opentelemetry_metrics.classify_error_type(None, status_code=499) == "CANCELLED" + ) + assert ( + _opentelemetry_metrics.classify_error_type(None, status_code=500) == "INTERNAL" + ) + assert ( + _opentelemetry_metrics.classify_error_type(None, status_code=501) + == "UNIMPLEMENTED" + ) + assert ( + _opentelemetry_metrics.classify_error_type(None, status_code=503) + == "UNAVAILABLE" + ) + assert ( + _opentelemetry_metrics.classify_error_type(None, status_code=504) + == "DEADLINE_EXCEEDED" + ) + assert _opentelemetry_metrics.classify_error_type(None, status_code=502) == "502" + assert _opentelemetry_metrics.classify_error_type(None, status_code=99999) == "OK" + + +def test_classify_error_type_exceptions(): + # 1. Checksum Mismatch (including subclasses of DataCorruption) + resp_mock = mock.Mock(status_code=200) + assert ( + _opentelemetry_metrics.classify_error_type( + storage_exceptions.DataCorruption(resp_mock, "crc mismatch") + ) + == "CHECKSUM_MISMATCH" + ) + + class CustomDataCorruption(storage_exceptions.DataCorruption): + pass + + assert ( + _opentelemetry_metrics.classify_error_type( + CustomDataCorruption(resp_mock, "subclass mismatch") + ) + == "CHECKSUM_MISMATCH" + ) + + # 2. Authentication Errors + assert ( + _opentelemetry_metrics.classify_error_type( + auth_exceptions.RefreshError("token expired") + ) + == "AUTHENTICATION_ERROR" + ) + assert ( + _opentelemetry_metrics.classify_error_type( + auth_exceptions.TransportError("auth transport error") + ) + == "AUTHENTICATION_ERROR" + ) + assert ( + _opentelemetry_metrics.classify_error_type( + auth_exceptions.DefaultCredentialsError("missing adc") + ) + == "AUTHENTICATION_ERROR" + ) + + # 3. Cancellation + assert ( + _opentelemetry_metrics.classify_error_type(asyncio.CancelledError()) + == "CANCELLED" + ) + + # 4. HTTP Conflict (409) consistency with status_code=409 + conflict_exc = api_exceptions.Conflict("conflict") + assert _opentelemetry_metrics.classify_error_type(conflict_exc) == "ABORTED" + + # 5. gRPC / GoogleAPICallError status codes + not_found_exc = api_exceptions.NotFound("not found") + assert _opentelemetry_metrics.classify_error_type(not_found_exc) == "NOT_FOUND" + + custom_api_exc = api_exceptions.GoogleAPICallError("custom") + custom_api_exc.grpc_status_code = "RAW_GRPC_STATUS" # type: ignore[assignment] + assert ( + _opentelemetry_metrics.classify_error_type(custom_api_exc) == "RAW_GRPC_STATUS" + ) + + # Direct grpc.RpcError-like exception with .code() method + class FakeStatusCode: + name = "DEADLINE_EXCEEDED" + + class FakeRpcError(Exception): + def code(self): + return FakeStatusCode() + + assert ( + _opentelemetry_metrics.classify_error_type(FakeRpcError()) + == "DEADLINE_EXCEEDED" + ) + + class FakeRpcErrorNoName(Exception): + def code(self): + return object() + + assert _opentelemetry_metrics.classify_error_type(FakeRpcErrorNoName()) == "UNKNOWN" + + # 6. HTTP status code on exception + http_api_exc = api_exceptions.GoogleAPICallError("bad gateway") + http_api_exc.code = 502 + http_api_exc.grpc_status_code = None + assert _opentelemetry_metrics.classify_error_type(http_api_exc) == "502" + + invalid_resp_with_code = storage_exceptions.InvalidResponse( + mock.Mock(status_code=429), "rate limited" + ) + assert ( + _opentelemetry_metrics.classify_error_type(invalid_resp_with_code) + == "RESOURCE_EXHAUSTED" + ) + + req_http_err = requests.exceptions.HTTPError(response=mock.Mock(status_code=503)) + assert _opentelemetry_metrics.classify_error_type(req_http_err) == "UNAVAILABLE" + + # 7. Timeouts + assert ( + _opentelemetry_metrics.classify_error_type( + requests.exceptions.ConnectTimeout("connect timeout") + ) + == "TIMEOUT" + ) + assert ( + _opentelemetry_metrics.classify_error_type( + requests.exceptions.ReadTimeout("read timeout") + ) + == "TIMEOUT" + ) + assert ( + _opentelemetry_metrics.classify_error_type(socket.timeout("timed out")) + == "TIMEOUT" + ) + assert ( + _opentelemetry_metrics.classify_error_type( + urllib3.exceptions.ReadTimeoutError(None, "/", "read timed out") # type: ignore[arg-type] + ) + == "TIMEOUT" + ) + assert ( + _opentelemetry_metrics.classify_error_type(concurrent.futures.TimeoutError()) + == "TIMEOUT" + ) + retry_err = api_exceptions.RetryError( + "deadline exceeded", cause=socket.timeout("timed out") + ) + assert _opentelemetry_metrics.classify_error_type(retry_err) == "TIMEOUT" + + # 8. TLS / SSL Failure + assert ( + _opentelemetry_metrics.classify_error_type( + requests.exceptions.SSLError("cert verify failed") + ) + == "TLS_FAILURE" + ) + assert ( + _opentelemetry_metrics.classify_error_type(ssl.SSLError("ssl handshake error")) + == "TLS_FAILURE" + ) + + # 9. DNS, Connection Refused, Connection Reset (direct and wrapped in requests/urllib3) + gai_err = socket.gaierror(-2, "Name or service not known") + new_conn_err = urllib3.exceptions.NewConnectionError(None, "dns failed") # type: ignore[arg-type] + new_conn_err.__cause__ = gai_err + wrapped_dns = requests.exceptions.ConnectionError( + urllib3.exceptions.MaxRetryError( + None, # type: ignore[arg-type] + "/", + reason=new_conn_err, + ) + ) + assert _opentelemetry_metrics.classify_error_type(wrapped_dns) == "DNS_FAILURE" + + refused_err = ConnectionRefusedError(111, "Connection refused") + wrapped_refused = requests.exceptions.ConnectionError(refused_err) + assert ( + _opentelemetry_metrics.classify_error_type(wrapped_refused) + == "CONNECTION_REFUSED" + ) + + reset_err = ConnectionResetError(104, "Connection reset by peer") + wrapped_reset = requests.exceptions.ConnectionError(reset_err) + assert ( + _opentelemetry_metrics.classify_error_type(wrapped_reset) == "CONNECTION_RESET" + ) + + # 10. Generic Connectivity + assert ( + _opentelemetry_metrics.classify_error_type( + requests.exceptions.ConnectionError("connection aborted") + ) + == "CONNECTIVITY" + ) + assert ( + _opentelemetry_metrics.classify_error_type( + requests.exceptions.ChunkedEncodingError("incomplete read") + ) + == "CONNECTIVITY" + ) + assert ( + _opentelemetry_metrics.classify_error_type(OSError("network unreachable")) + == "CONNECTIVITY" + ) + assert ( + _opentelemetry_metrics.classify_error_type( + http.client.IncompleteRead(b"partial", 10) + ) + == "CONNECTIVITY" + ) + + # 11. Explicit status_code fallback when exception is generic + assert ( + _opentelemetry_metrics.classify_error_type( + RuntimeError("unexpected"), status_code=503 + ) + == "UNAVAILABLE" + ) + + # 12. Client Validation Errors + assert ( + _opentelemetry_metrics.classify_error_type(ValueError("bad arg")) + == "CLIENT_VALIDATION_ERROR" + ) + assert ( + _opentelemetry_metrics.classify_error_type(TypeError("bad type")) + == "CLIENT_VALIDATION_ERROR" + ) + invalid_resp_no_code = storage_exceptions.InvalidResponse( + mock.Mock(status_code=None), "missing header" + ) + assert ( + _opentelemetry_metrics.classify_error_type(invalid_resp_no_code) + == "CLIENT_VALIDATION_ERROR" + ) + + # 13. Cyclic exception chain and unknown exception + e1 = RuntimeError("cycle 1") + e2 = RuntimeError("cycle 2", e1) + e1.__cause__ = e2 + assert _opentelemetry_metrics.classify_error_type(e1) == "UNKNOWN" + + +# --------------------------------------------------------------------------- +# Route Normalization (`normalize_url_template`) Tests +# --------------------------------------------------------------------------- +@pytest.mark.parametrize( + "raw_url,expected", + [ + (None, "UNKNOWN"), + ("", "UNKNOWN"), + (123, "UNKNOWN"), + ("https://storage.googleapis.com", "UNKNOWN"), + ("/", "UNKNOWN"), + ("///", "UNKNOWN"), + ( + "https://storage.googleapis.com//storage/v1/b/my-bucket//", + "/storage/v1/b/{bucket}", + ), + ( + "https://storage.googleapis.com/upload/storage/v1/b/my-bucket/o/my-obj?uploadType=resumable", + "/upload/storage/v1/b/{bucket}/o", + ), + ( + "https://storage.googleapis.com/custom/path?tag=my_uploads", + "UNKNOWN", + ), + ( + "https://storage.googleapis.com/custom/path?my_uploadId=123", + "UNKNOWN", + ), + ( + "https://storage.googleapis.com/storage/v1/b?project=my-proj", + "/storage/v1/b", + ), + ("/b?project=my-proj", "/storage/v1/b"), + ( + "https://storage.googleapis.com/storage/v1/b/my-bucket?projection=full", + "/storage/v1/b/{bucket}", + ), + ("/b/my-bucket/", "/storage/v1/b/{bucket}"), + ("b/my-bucket", "/storage/v1/b/{bucket}"), + ( + "https://storage.googleapis.com/storage/v1/b/my-bucket/o?prefix=foo", + "/storage/v1/b/{bucket}/o", + ), + ("/b/my-bucket/o", "/storage/v1/b/{bucket}/o"), + ( + "https://storage.googleapis.com/storage/v1/b/my-bucket/o/folder%2Ffile.txt?generation=1", + "/storage/v1/b/{bucket}/o/{object}", + ), + ("/b/my-bucket/o/file.txt", "/storage/v1/b/{bucket}/o/{object}"), + ( + "https://storage.googleapis.com/storage/v1/b/my-bucket/o/dest.txt/compose", + "/storage/v1/b/{bucket}/o/{object}/compose", + ), + ( + "/b/my-bucket/o/dest.txt/compose", + "/storage/v1/b/{bucket}/o/{object}/compose", + ), + ( + "https://storage.googleapis.com/storage/v1/b/src-b/o/src-o/rewriteTo/b/dst-b/o/dst-o", + "/storage/v1/b/{bucket}/o/{object}/rewriteTo/b/{bucket}/o/{object}", + ), + ( + "/b/src-b/o/src-o/rewriteTo/b/dst-b/o/dst-o", + "/storage/v1/b/{bucket}/o/{object}/rewriteTo/b/{bucket}/o/{object}", + ), + ( + "https://storage.googleapis.com/storage/v1/b/src-b/o/src-o/copyTo/b/dst-b/o/dst-o", + "/storage/v1/b/{bucket}/o/{object}/copyTo/b/{bucket}/o/{object}", + ), + ( + "/b/src-b/o/src-o/copyTo/b/dst-b/o/dst-o", + "/storage/v1/b/{bucket}/o/{object}/copyTo/b/{bucket}/o/{object}", + ), + ( + "https://storage.googleapis.com/storage/v1/b/my-bucket/o/src-o/moveTo/o/dst-o", + "/storage/v1/b/{bucket}/o/{object}/moveTo/o/{object}", + ), + ( + "/b/my-bucket/o/src-o/moveTo/o/dst-o", + "/storage/v1/b/{bucket}/o/{object}/moveTo/o/{object}", + ), + ( + "https://storage.googleapis.com/storage/v1/b/my-bucket/o/my-o/restore", + "/storage/v1/b/{bucket}/o/{object}/restore", + ), + ( + "/b/my-bucket/o/my-o/restore", + "/storage/v1/b/{bucket}/o/{object}/restore", + ), + ( + "/b/my-bucket/o/my-o/acl", + "/storage/v1/b/{bucket}/o/{object}/acl", + ), + ( + "/b/my-bucket/o/my-o/acl/allUsers", + "/storage/v1/b/{bucket}/o/{object}/acl/{entity}", + ), + ( + "/b/my-bucket/o/my-o/iam", + "/storage/v1/b/{bucket}/o/{object}/iam", + ), + ( + "/b/my-bucket/o/my-o/iam/testPermissions", + "/storage/v1/b/{bucket}/o/{object}/iam", + ), + ( + "/b/my-bucket/acl", + "/storage/v1/b/{bucket}/acl", + ), + ( + "/b/my-bucket/acl/allUsers", + "/storage/v1/b/{bucket}/acl/{entity}", + ), + ( + "/b/my-bucket/defaultObjectAcl", + "/storage/v1/b/{bucket}/defaultObjectAcl", + ), + ( + "/b/my-bucket/defaultObjectAcl/allUsers", + "/storage/v1/b/{bucket}/defaultObjectAcl/{entity}", + ), + ( + "/b/my-bucket/restore", + "/storage/v1/b/{bucket}/restore", + ), + ( + "https://storage.googleapis.com/storage/v1/b/my-bucket/iam/testPermissions", + "/storage/v1/b/{bucket}/iam", + ), + ("/b/my-bucket/iam", "/storage/v1/b/{bucket}/iam"), + ( + "https://storage.googleapis.com/storage/v1/b/my-bucket/notificationConfigs", + "/storage/v1/b/{bucket}/notificationConfigs", + ), + ( + "/b/my-bucket/notificationConfigs", + "/storage/v1/b/{bucket}/notificationConfigs", + ), + ( + "https://storage.googleapis.com/storage/v1/b/my-bucket/notificationConfigs/42", + "/storage/v1/b/{bucket}/notificationConfigs/{notification}", + ), + ( + "/b/my-bucket/notificationConfigs/42", + "/storage/v1/b/{bucket}/notificationConfigs/{notification}", + ), + ( + "https://storage.googleapis.com/storage/v1/b/my-bucket/lockRetentionPolicy", + "/storage/v1/b/{bucket}/lockRetentionPolicy", + ), + ( + "/b/my-bucket/lockRetentionPolicy", + "/storage/v1/b/{bucket}/lockRetentionPolicy", + ), + ( + "/storage/v1/projects/my-proj/serviceAccount", + "/storage/v1/projects/{project}/serviceAccount", + ), + ( + "/projects/my-proj/serviceAccount", + "/storage/v1/projects/{project}/serviceAccount", + ), + ( + "/storage/v1/projects/my-proj/hmacKeys", + "/storage/v1/projects/{project}/hmacKeys", + ), + ("/projects/my-proj/hmacKeys", "/storage/v1/projects/{project}/hmacKeys"), + ( + "/storage/v1/projects/my-proj/hmacKeys/GOOG123", + "/storage/v1/projects/{project}/hmacKeys/{access_id}", + ), + ( + "/projects/my-proj/hmacKeys/GOOG123", + "/storage/v1/projects/{project}/hmacKeys/{access_id}", + ), + ( + "https://storage.googleapis.com/upload/storage/v1/b/my-bucket/o?uploadType=resumable", + "/upload/storage/v1/b/{bucket}/o", + ), + ( + "https://storage.googleapis.com/download/storage/v1/b/my-bucket/o/my-obj?alt=media", + "/download/storage/v1/b/{bucket}/o/{object}", + ), + ("https://storage.googleapis.com/batch/storage/v1", "/batch/storage/v1"), + ( + "https://storage.googleapis.com/my-bucket/my-object?uploads", + "/{bucket}/{object}", + ), + ( + "https://storage.googleapis.com/my-bucket/nested/dir/object.bin?uploadId=xyz&partNumber=2", + "/{bucket}/{object}", + ), + ( + "https://storage.googleapis.com/unrecognized/two-segment", + "UNKNOWN", + ), + ("https://storage.googleapis.com/storage/v2", "UNKNOWN"), + ("https://storage.googleapis.com/storage/v2/other?uploadId=xyz", "UNKNOWN"), + ("https://storage.googleapis.com/custom/unrecognized/path/here", "UNKNOWN"), + ], +) +def test_normalize_url_template(raw_url, expected): + assert _opentelemetry_metrics.normalize_url_template(raw_url) == expected + + +# --------------------------------------------------------------------------- +# Logical Operation Context (`start_operation`) Tests +# --------------------------------------------------------------------------- +def test_start_operation_disabled_and_nesting(monkeypatch): + monkeypatch.setattr(_opentelemetry_metrics, "_ENABLE_METRICS_DEV_GATE", True) + mock_metrics = mock.Mock() + mock_meter = mock.Mock() + mock_meter.create_histogram.side_effect = lambda *a, **kw: mock.Mock() + mock_meter.create_counter.side_effect = lambda *a, **kw: mock.Mock() + mock_metrics.get_meter.return_value = mock_meter + monkeypatch.setattr(_opentelemetry_metrics, "metrics", mock_metrics, raising=False) + + # Disabled recorder -> yields None and records nothing + disabled_recorder = _opentelemetry_metrics._MetricsRecorder(enable_metrics=False) + with _opentelemetry_metrics.start_operation( + disabled_recorder, "storage.Blob.reload", "storage.objects.get" + ) as op: + assert op is None + assert _opentelemetry_metrics.get_current_operation() is None + + # Enabled recorder -> outermost call records, nested call is suppressed + enabled_recorder = _opentelemetry_metrics._MetricsRecorder(enable_metrics=True) + with _opentelemetry_metrics.start_operation( + enabled_recorder, + "storage.Blob.download_to_filename", + "storage.objects.get", + ) as outer_op: + assert outer_op is not None + assert outer_op.client_method == "storage.Blob.download_to_filename" + assert outer_op.rpc_method == "storage.objects.get" + assert _opentelemetry_metrics.get_current_operation() == outer_op + + with _opentelemetry_metrics.start_operation( + enabled_recorder, + "storage.Blob.download_to_file", + "storage.objects.get", + ) as inner_op: + assert inner_op is outer_op + + assert _opentelemetry_metrics.get_current_operation() is None + enabled_recorder.client_request_duration.record.assert_called_once() + enabled_recorder.operations_counter.add.assert_called_once() + _, kwargs = enabled_recorder.operations_counter.add.call_args + assert ( + kwargs["attributes"]["gcp.client.method"] == "storage.Blob.download_to_filename" + ) + assert kwargs["attributes"]["gcp.storage.operation"] == "storage.objects.get" + assert kwargs["attributes"]["error.type"] == "OK" + + +def test_start_operation_exception_and_suppression(monkeypatch): + monkeypatch.setattr(_opentelemetry_metrics, "_ENABLE_METRICS_DEV_GATE", True) + mock_metrics = mock.Mock() + mock_meter = mock.Mock() + mock_meter.create_histogram.side_effect = lambda *a, **kw: mock.Mock() + mock_meter.create_counter.side_effect = lambda *a, **kw: mock.Mock() + mock_metrics.get_meter.return_value = mock_meter + monkeypatch.setattr(_opentelemetry_metrics, "metrics", mock_metrics, raising=False) + + recorder = _opentelemetry_metrics._MetricsRecorder(enable_metrics=True) + + with pytest.raises(api_exceptions.NotFound): + with _opentelemetry_metrics.start_operation( + recorder, "storage.Bucket.get_blob", "storage.objects.get" + ): + raise api_exceptions.NotFound("missing") + + assert _opentelemetry_metrics.get_current_operation() is None + _, kwargs = recorder.operations_counter.add.call_args + assert kwargs["attributes"]["error.type"] == "NOT_FOUND" + + # Suppressed outer operation (e.g. transfer_manager worker) suppresses both + # logical operation metrics and body size recording + recorder.operations_counter.add.reset_mock() + recorder.client_request_duration.record.reset_mock() + with _opentelemetry_metrics.start_operation( + None, + "storage.transfer_manager.download_many_to_path", + "storage.objects.get", + suppress=True, + ) as suppressed_op: + assert suppressed_op is not None + assert suppressed_op.suppressed is True + with _opentelemetry_metrics.start_operation( + recorder, + "storage.Blob.download_to_filename", + "storage.objects.get", + ): + recorder.record_request_body_size(2048) + recorder.record_response_body_size(4096) + + assert _opentelemetry_metrics.get_current_operation() is None + recorder.operations_counter.add.assert_not_called() + recorder.client_request_duration.record.assert_not_called() + recorder.request_body_size.record.assert_not_called() + recorder.response_body_size.record.assert_not_called() + + # Nested suppress=True inside an active unsuppressed operation (e.g. BlobReader prefetch) + with _opentelemetry_metrics.start_operation( + recorder, + "storage.Blob.open", + "storage.objects.get", + ) as active_op: + assert active_op is not None and active_op.suppressed is False + with _opentelemetry_metrics.start_operation( + recorder, + "storage.Blob.download_as_bytes", + "storage.objects.get", + suppress=True, + ) as child_suppressed: + assert child_suppressed is not None + assert child_suppressed.suppressed is True + assert child_suppressed.client_method == "storage.Blob.open" + # Prefetch body size and TTFB are suppressed inside child_suppressed + recorder.record_response_body_size(40 * 1024 * 1024) + recorder.record_ttfb(0.05) + # Outside the suppressed prefetch block, actual user bytes are recorded + assert _opentelemetry_metrics.get_current_operation() == active_op + recorder.record_response_body_size(1024) + recorder.record_response_body_size(True) # type: ignore[arg-type] + recorder.record_request_body_size(False) # type: ignore[arg-type] + + recorder.response_body_size.record.assert_called_once() + assert recorder.response_body_size.record.call_args[0][0] == 1024 + recorder.operation_ttfb.record.assert_not_called() + recorder.operations_counter.add.assert_called_once() + + +def test_start_operation_asyncio_concurrency_isolation(monkeypatch): + monkeypatch.setattr(_opentelemetry_metrics, "_ENABLE_METRICS_DEV_GATE", True) + mock_metrics = mock.Mock() + mock_meter = mock.Mock() + mock_meter.create_histogram.side_effect = lambda *a, **kw: mock.Mock() + mock_meter.create_counter.side_effect = lambda *a, **kw: mock.Mock() + mock_metrics.get_meter.return_value = mock_meter + monkeypatch.setattr(_opentelemetry_metrics, "metrics", mock_metrics, raising=False) + + recorder = _opentelemetry_metrics._MetricsRecorder( + enable_metrics=True, rpc_system="grpc" + ) + + async def worker(method_name: str, rpc_name: str): + with _opentelemetry_metrics.start_operation(recorder, method_name, rpc_name): + await asyncio.sleep(0.005) + current = _opentelemetry_metrics.get_current_operation() + assert current is not None + assert current.client_method == method_name + assert current.rpc_method == rpc_name + + async def run_all(): + await asyncio.gather( + worker( + "storage.AsyncMultiRangeDownloader.download_ranges", + "google.storage.v2.Storage/BidiReadObject", + ), + worker( + "storage.AsyncAppendableObjectWriter.append", + "google.storage.v2.Storage/BidiWriteObject", + ), + ) + + asyncio.run(run_all()) + assert recorder.operations_counter.add.call_count == 2 + + +# --------------------------------------------------------------------------- +# `_extract_server_address` & `_MetricsRecorder` Tests +# --------------------------------------------------------------------------- +@pytest.mark.parametrize( + "endpoint,expected", + [ + (None, "storage.googleapis.com"), + ("", "storage.googleapis.com"), + (" ", "storage.googleapis.com"), + ("https://storage.googleapis.com:443", "storage.googleapis.com"), + ("storage.googleapis.com:443", "storage.googleapis.com"), + ("//storage.googleapis.com:443", "storage.googleapis.com"), + ("https://[::1]:8080", "::1"), + ("http://localhost:9023/storage/v1", "localhost"), + ("https://", "storage.googleapis.com"), + ], +) +def test_extract_server_address(endpoint, expected): + assert _opentelemetry_metrics._extract_server_address(endpoint) == expected + + +def test_metrics_recorder_recording_and_shutdown(monkeypatch): + monkeypatch.setattr(_opentelemetry_metrics, "_ENABLE_METRICS_DEV_GATE", True) + mock_metrics = mock.Mock() + mock_meter = mock.Mock() + mock_meter.create_histogram.side_effect = lambda *a, **kw: mock.Mock() + mock_meter.create_counter.side_effect = lambda *a, **kw: mock.Mock() + mock_metrics.get_meter.return_value = mock_meter + monkeypatch.setattr(_opentelemetry_metrics, "metrics", mock_metrics, raising=False) + + mock_provider = mock.Mock() + recorder = _opentelemetry_metrics._MetricsRecorder( + enable_metrics=True, + meter_provider=mock_provider, + api_endpoint="https://custom.storage.googleapis.com:443", + rpc_system="http", + ) + assert recorder.standard_enabled is True + assert recorder.enabled is True + assert recorder.server_address == "custom.storage.googleapis.com" + + base_attrs = recorder.get_base_attributes() + assert base_attrs["server.address"] == "custom.storage.googleapis.com" + assert base_attrs["rpc.system.name"] == "http" + + # Verify exact instrument names, units, and descriptions + histogram_specs = { + call.args[0]: (call.kwargs["unit"], call.kwargs["description"]) + for call in mock_meter.create_histogram.call_args_list + } + assert histogram_specs == { + "rpc.client.call.duration": ( + "s", + "Duration of an individual gRPC attempt in seconds (excluding retries).", + ), + "http.client.request.duration": ( + "s", + "Duration of an individual HTTP request attempt in seconds (excluding retries).", + ), + "gcp.client.request.duration": ( + "s", + "End-to-end latency of a logical user-facing Veneer operation in seconds.", + ), + "gcp.storage.client.request.body.size": ( + "By", + "Application payload bytes submitted by the caller for an upload or write operation.", + ), + "gcp.storage.client.response.body.size": ( + "By", + "Application payload bytes delivered to the caller.", + ), + "gcp.storage.client.operation.ttfb": ( + "s", + "Time elapsed from attempt initiation to receipt of the first response byte.", + ), + } + counter_specs = { + call.args[0]: (call.kwargs["unit"], call.kwargs["description"]) + for call in mock_meter.create_counter.call_args_list + } + assert counter_specs == { + "gcp.storage.client.operations": ( + "{operation}", + "Number of logical GCS client operations.", + ), + "gcp.storage.client.attempts": ( + "{attempt}", + "Number of physical network transport attempts executed.", + ), + "gcp.storage.client.errors": ( + "{error}", + "Number of physical network transport attempts terminating with a non-OK status.", + ), + } + + # Record within a logical operation context + with _opentelemetry_metrics.start_operation( + recorder, "storage.Blob.upload_from_string", "storage.objects.insert" + ): + recorder.record_http_attempt( + duration=0.05, + http_method="post", + url_template="/upload/storage/v1/b/{bucket}/o", + status_code=503, + error_type="UNAVAILABLE", + network_path="cloudpath", + ) + recorder.record_http_attempt( + duration=0.04, + http_method="POST", + url_template="/upload/storage/v1/b/{bucket}/o", + status_code=200, + error_type="OK", + ) + recorder.record_request_body_size(1024) + recorder.record_response_body_size(512) + recorder.record_ttfb(0.01) + + assert recorder.http_request_duration.record.call_count == 2 + _, http_dur_kwargs = recorder.http_request_duration.record.call_args_list[0] + assert http_dur_kwargs["attributes"]["http.request.method"] == "POST" + assert ( + http_dur_kwargs["attributes"]["url.template"] + == "/upload/storage/v1/b/{bucket}/o" + ) + assert http_dur_kwargs["attributes"]["http.response.status_code"] == 503 + assert http_dur_kwargs["attributes"]["error.type"] == "UNAVAILABLE" + assert http_dur_kwargs["attributes"]["network.path"] == "cloudpath" + assert "gcp.storage.operation" not in http_dur_kwargs["attributes"] + assert "gcp.client.method" not in http_dur_kwargs["attributes"] + + assert recorder.attempts_counter.add.call_count == 2 + _, attempt_kwargs = recorder.attempts_counter.add.call_args_list[0] + assert ( + attempt_kwargs["attributes"]["gcp.storage.operation"] + == "storage.objects.insert" + ) + assert attempt_kwargs["attributes"]["http.response.status_code"] == 503 + assert attempt_kwargs["attributes"]["error.type"] == "UNAVAILABLE" + assert attempt_kwargs["attributes"]["network.path"] == "cloudpath" + assert "url.template" not in attempt_kwargs["attributes"] + assert "http.request.method" not in attempt_kwargs["attributes"] + + # Only the 503 attempt records an error + recorder.errors_counter.add.assert_called_once() + _, err_kwargs = recorder.errors_counter.add.call_args + assert err_kwargs["attributes"]["error.type"] == "UNAVAILABLE" + assert err_kwargs["attributes"]["gcp.storage.operation"] == "storage.objects.insert" + assert ( + err_kwargs["attributes"]["gcp.client.method"] + == "storage.Blob.upload_from_string" + ) + assert "url.template" not in err_kwargs["attributes"] + assert "http.response.status_code" not in err_kwargs["attributes"] + + recorder.request_body_size.record.assert_called_once() + recorder.response_body_size.record.assert_called_once() + recorder.operation_ttfb.record.assert_called_once() + + # gRPC attempt recording (OK and non-OK) + recorder.errors_counter.add.reset_mock() + recorder.attempts_counter.add.reset_mock() + recorder.record_rpc_attempt( + duration=0.02, + operation="google.storage.v2.Storage/BidiReadObject", + status_code="OK", + error_type="OK", + client_method="storage.AsyncMultiRangeDownloader.download_ranges", + network_path="directpath", + ) + recorder.record_rpc_attempt( + duration=0.03, + operation="google.storage.v2.Storage/BidiReadObject", + status_code="UNAVAILABLE", + error_type="UNAVAILABLE", + client_method="storage.AsyncMultiRangeDownloader.download_ranges", + ) + assert recorder.rpc_call_duration.record.call_count == 2 + _, rpc_dur_kwargs = recorder.rpc_call_duration.record.call_args_list[0] + assert ( + rpc_dur_kwargs["attributes"]["gcp.storage.operation"] + == "google.storage.v2.Storage/BidiReadObject" + ) + assert ( + rpc_dur_kwargs["attributes"]["gcp.client.method"] + == "storage.AsyncMultiRangeDownloader.download_ranges" + ) + assert rpc_dur_kwargs["attributes"]["rpc.response.status_code"] == "OK" + assert rpc_dur_kwargs["attributes"]["network.path"] == "directpath" + + assert recorder.attempts_counter.add.call_count == 2 + _, rpc_attempt_kwargs = recorder.attempts_counter.add.call_args_list[0] + assert ( + rpc_attempt_kwargs["attributes"]["gcp.storage.operation"] + == "google.storage.v2.Storage/BidiReadObject" + ) + assert rpc_attempt_kwargs["attributes"]["rpc.response.status_code"] == "OK" + assert rpc_attempt_kwargs["attributes"]["error.type"] == "OK" + assert rpc_attempt_kwargs["attributes"]["network.path"] == "directpath" + assert "gcp.client.method" not in rpc_attempt_kwargs["attributes"] + + recorder.errors_counter.add.assert_called_once() + _, rpc_err_kwargs = recorder.errors_counter.add.call_args + assert rpc_err_kwargs["attributes"]["error.type"] == "UNAVAILABLE" + assert "rpc.response.status_code" not in rpc_err_kwargs["attributes"] + + # Shutdown flushes custom meter_provider and swallows force_flush errors gracefully + recorder.shutdown() + mock_provider.force_flush.assert_called_once() + + mock_provider.force_flush.side_effect = RuntimeError("flush failed") + recorder.shutdown() # Must not raise + + +def test_metrics_recorder_disabled_no_ops(monkeypatch): + monkeypatch.setattr(_opentelemetry_metrics, "_ENABLE_METRICS_DEV_GATE", False) + recorder = _opentelemetry_metrics._MetricsRecorder(enable_metrics=True) + assert recorder.standard_enabled is False + assert recorder.enabled is False + + # All recording methods are safe no-ops when disabled + recorder.record_logical_operation(0.1, "m", "op") + recorder.record_http_attempt(0.1, "GET", "/storage/v1/b") + recorder.record_rpc_attempt(0.1, "google.storage.v2.Storage/ReadObject") + recorder.record_request_body_size(100) + recorder.record_response_body_size(100) + recorder.record_ttfb(0.01) + recorder.shutdown() From 799ac3a81cd9c49b633cbb60516395c1158b83ed Mon Sep 17 00:00:00 2001 From: Shradha Katyal Date: Mon, 19 Oct 2026 19:02:17 +0000 Subject: [PATCH 2/3] fix(storage): fix ruff lint, py310 lower-bound metrics fallback, and branch coverage [Generated-by: AI] --- .../cloud/storage/_opentelemetry_metrics.py | 47 ++++++++---- .../tests/unit/test__opentelemetry_metrics.py | 74 +++++++++++++++++-- 2 files changed, 98 insertions(+), 23 deletions(-) diff --git a/packages/google-cloud-storage/google/cloud/storage/_opentelemetry_metrics.py b/packages/google-cloud-storage/google/cloud/storage/_opentelemetry_metrics.py index c357a968407f..f7394c9b2740 100644 --- a/packages/google-cloud-storage/google/cloud/storage/_opentelemetry_metrics.py +++ b/packages/google-cloud-storage/google/cloud/storage/_opentelemetry_metrics.py @@ -15,11 +15,8 @@ """Manages OpenTelemetry metrics instruments and gating for GCS client.""" import asyncio -from collections import deque import concurrent.futures import contextlib -from contextvars import ContextVar -from dataclasses import dataclass import http.client import logging import os @@ -27,6 +24,9 @@ import socket import ssl import time +from collections import deque +from contextvars import ContextVar +from dataclasses import dataclass from typing import Any, Deque, Dict, Iterator, List, Optional, Set, Tuple from urllib.parse import urlsplit @@ -85,6 +85,7 @@ HAS_OPENTELEMETRY_METRICS = True except ImportError: + metrics = None # type: ignore[assignment] HAS_OPENTELEMETRY_METRICS = False logger.debug( "OpenTelemetry metrics package (opentelemetry-api >= 1.12.0) is not " @@ -207,9 +208,12 @@ def get_meter( Optional[Any]: The OpenTelemetry Meter instance if standard or debug metrics are enabled, or None otherwise. """ - if not ( - is_metrics_enabled(enable_metrics) - or is_debug_metrics_enabled(enable_debug_metrics) + if ( + not ( + is_metrics_enabled(enable_metrics) + or is_debug_metrics_enabled(enable_debug_metrics) + ) + or metrics is None ): return None return metrics.get_meter( @@ -238,13 +242,13 @@ def _build_latency_buckets() -> Tuple[float, ...]: boundary += increment increment = 0.010 - for i in range(150): - if boundary >= 300.0: - break + i = 0 + while boundary < 300.0: if i != 0 and i % 10 == 0: increment *= 2 boundaries.append(round(boundary, 6)) boundary += increment + i += 1 return tuple(boundaries) @@ -452,11 +456,10 @@ def classify_error_type( if callable(code_attr): try: rpc_status = code_attr() - if rpc_status is not None: - name = getattr(rpc_status, "name", None) - if name: - return str(name) - except Exception: # pragma: no cover + name = getattr(rpc_status, "name", None) + if name: + return str(name) + except Exception: # pragma: NO COVER pass if _is_valid_status_code(code_attr): @@ -777,7 +780,9 @@ def start_operation( yield existing return - recorder_active = recorder is not None and recorder.standard_enabled + recorder_active = ( + recorder is not None and getattr(recorder, "standard_enabled", False) is True + ) if not recorder_active and not suppress: yield None return @@ -807,6 +812,14 @@ def start_operation( ) +def get_active_recorder(obj: Any) -> Optional["_MetricsRecorder"]: + """Returns ``obj._metrics_recorder`` if it is an active ``_MetricsRecorder``.""" + recorder = getattr(obj, "_metrics_recorder", None) + if recorder is not None and getattr(recorder, "standard_enabled", False) is True: + return recorder + return None + + # --------------------------------------------------------------------------- # 10. Server Address Parsing & `_MetricsRecorder` # --------------------------------------------------------------------------- @@ -817,7 +830,9 @@ def _extract_server_address(api_endpoint: Optional[str]) -> str: endpoint = api_endpoint.strip() if not endpoint: return DEFAULT_SERVER_ADDRESS - if "://" not in endpoint: + if ":///" in endpoint: + endpoint = "https://" + endpoint.split(":///", 1)[1].lstrip("/") + elif "://" not in endpoint: endpoint = "https://" + endpoint.lstrip("/") parsed = urlsplit(endpoint) return parsed.hostname or DEFAULT_SERVER_ADDRESS diff --git a/packages/google-cloud-storage/tests/unit/test__opentelemetry_metrics.py b/packages/google-cloud-storage/tests/unit/test__opentelemetry_metrics.py index cb392b2f9870..275baa8f7c5c 100644 --- a/packages/google-cloud-storage/tests/unit/test__opentelemetry_metrics.py +++ b/packages/google-cloud-storage/tests/unit/test__opentelemetry_metrics.py @@ -22,11 +22,11 @@ import types from unittest import mock -from google.api_core import exceptions as api_exceptions -from google.auth import exceptions as auth_exceptions import pytest import requests.exceptions import urllib3.exceptions +from google.api_core import exceptions as api_exceptions +from google.auth import exceptions as auth_exceptions from google.cloud.storage import _opentelemetry_metrics from google.cloud.storage import exceptions as storage_exceptions @@ -57,6 +57,8 @@ def test_opentelemetry_import_error_on_load(monkeypatch): is False ) assert _opentelemetry_metrics.get_meter(enable_metrics=True) is None + m.setattr(_opentelemetry_metrics, "HAS_OPENTELEMETRY_METRICS", True) + assert _opentelemetry_metrics.get_meter(enable_metrics=True) is None finally: importlib.reload(_opentelemetry_metrics) @@ -1025,11 +1027,11 @@ async def worker(method_name: str, rpc_name: str): async def run_all(): await asyncio.gather( worker( - "storage.AsyncMultiRangeDownloader.download_ranges", + "storage.asyncio.AsyncMultiRangeDownloader.download_ranges", "google.storage.v2.Storage/BidiReadObject", ), worker( - "storage.AsyncAppendableObjectWriter.append", + "storage.asyncio.AsyncAppendableObjectWriter.append", "google.storage.v2.Storage/BidiWriteObject", ), ) @@ -1205,7 +1207,7 @@ def test_metrics_recorder_recording_and_shutdown(monkeypatch): operation="google.storage.v2.Storage/BidiReadObject", status_code="OK", error_type="OK", - client_method="storage.AsyncMultiRangeDownloader.download_ranges", + client_method="storage.asyncio.AsyncMultiRangeDownloader.download_ranges", network_path="directpath", ) recorder.record_rpc_attempt( @@ -1213,7 +1215,7 @@ def test_metrics_recorder_recording_and_shutdown(monkeypatch): operation="google.storage.v2.Storage/BidiReadObject", status_code="UNAVAILABLE", error_type="UNAVAILABLE", - client_method="storage.AsyncMultiRangeDownloader.download_ranges", + client_method="storage.asyncio.AsyncMultiRangeDownloader.download_ranges", ) assert recorder.rpc_call_duration.record.call_count == 2 _, rpc_dur_kwargs = recorder.rpc_call_duration.record.call_args_list[0] @@ -1223,7 +1225,7 @@ def test_metrics_recorder_recording_and_shutdown(monkeypatch): ) assert ( rpc_dur_kwargs["attributes"]["gcp.client.method"] - == "storage.AsyncMultiRangeDownloader.download_ranges" + == "storage.asyncio.AsyncMultiRangeDownloader.download_ranges" ) assert rpc_dur_kwargs["attributes"]["rpc.response.status_code"] == "OK" assert rpc_dur_kwargs["attributes"]["network.path"] == "directpath" @@ -1244,6 +1246,34 @@ def test_metrics_recorder_recording_and_shutdown(monkeypatch): assert rpc_err_kwargs["attributes"]["error.type"] == "UNAVAILABLE" assert "rpc.response.status_code" not in rpc_err_kwargs["attributes"] + # Record outside of start_operation with status_code=None and callable .code() returning None + recorder.record_http_attempt( + duration=0.01, + http_method="GET", + url_template="/storage/v1/b", + status_code=None, + error_type="UNKNOWN", + ) + recorder.record_rpc_attempt( + duration=0.01, + operation="google.storage.v2.Storage/GetObject", + status_code=None, + error_type="UNKNOWN", + ) + recorder.record_request_body_size(64) + + class _NoneCodeErr(Exception): + def code(self): + return None + + assert _opentelemetry_metrics.classify_error_type(_NoneCodeErr()) == "UNKNOWN" + assert ( + _opentelemetry_metrics.classify_error_type( + urllib3.exceptions.NewConnectionError(None, "conn err") + ) + == "CONNECTIVITY" + ) + # Shutdown flushes custom meter_provider and swallows force_flush errors gracefully recorder.shutdown() mock_provider.force_flush.assert_called_once() @@ -1252,6 +1282,36 @@ def test_metrics_recorder_recording_and_shutdown(monkeypatch): recorder.shutdown() # Must not raise +def test_get_active_recorder_and_grpc_uri_schemes(monkeypatch): + monkeypatch.setattr(_opentelemetry_metrics, "HAS_OPENTELEMETRY_METRICS", True) + monkeypatch.setattr(_opentelemetry_metrics, "_ENABLE_METRICS_DEV_GATE", True) + + assert ( + _opentelemetry_metrics._extract_server_address("dns:///custom.endpoint.com:443") + == "custom.endpoint.com" + ) + assert ( + _opentelemetry_metrics._extract_server_address( + "google-c2p-experimental:///storage.googleapis.com" + ) + == "storage.googleapis.com" + ) + + # Unconfigured MagicMock should not be treated as an active recorder + unconfigured_mock = mock.MagicMock() + assert _opentelemetry_metrics.get_active_recorder(unconfigured_mock) is None + assert _opentelemetry_metrics.get_active_recorder(None) is None + + active_rec = _opentelemetry_metrics._MetricsRecorder(enable_metrics=True) + holder = mock.MagicMock() + holder._metrics_recorder = active_rec + assert _opentelemetry_metrics.get_active_recorder(holder) is active_rec + + disabled_rec = _opentelemetry_metrics._MetricsRecorder(enable_metrics=False) + holder._metrics_recorder = disabled_rec + assert _opentelemetry_metrics.get_active_recorder(holder) is None + + def test_metrics_recorder_disabled_no_ops(monkeypatch): monkeypatch.setattr(_opentelemetry_metrics, "_ENABLE_METRICS_DEV_GATE", False) recorder = _opentelemetry_metrics._MetricsRecorder(enable_metrics=True) From 6544cdcf9bfe9190ad814eb1c639f3ce68eab2ed Mon Sep 17 00:00:00 2001 From: Shradha Katyal Date: Mon, 19 Oct 2026 19:36:20 +0000 Subject: [PATCH 3/3] fix(storage): harden error classification against mocks and future cancellations [Generated-by: AI] --- .../cloud/storage/_opentelemetry_metrics.py | 32 +++++++++++-------- .../tests/unit/test__opentelemetry_metrics.py | 14 ++++++++ 2 files changed, 33 insertions(+), 13 deletions(-) diff --git a/packages/google-cloud-storage/google/cloud/storage/_opentelemetry_metrics.py b/packages/google-cloud-storage/google/cloud/storage/_opentelemetry_metrics.py index f7394c9b2740..27675bc298a0 100644 --- a/packages/google-cloud-storage/google/cloud/storage/_opentelemetry_metrics.py +++ b/packages/google-cloud-storage/google/cloud/storage/_opentelemetry_metrics.py @@ -362,21 +362,25 @@ def _iter_exception_chain(exc: BaseException) -> Iterator[BaseException]: visited.add(id(nested)) queue.append(nested) - 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) def _is_subclass_by_module( - cls: type, module_prefix: str, class_names: Optional[Set[str]] = None + cls: type, module_prefix: str, class_names: Set[str] ) -> bool: """Checks whether ``cls`` or any of its base classes match module/name criteria.""" - 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 + 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 getattr(base, "__name__", None) in class_names: + return True return False @@ -426,7 +430,9 @@ def classify_error_type( # 3. Cancellation for item in chain: - if isinstance(item, asyncio.CancelledError): + if isinstance( + item, (asyncio.CancelledError, concurrent.futures.CancelledError) + ): return "CANCELLED" # 4. HTTP response status code on exception (checked before grpc_status_code so @@ -457,8 +463,8 @@ def classify_error_type( try: rpc_status = code_attr() name = getattr(rpc_status, "name", None) - if name: - return str(name) + if isinstance(name, str) and name: + return name except Exception: # pragma: NO COVER pass diff --git a/packages/google-cloud-storage/tests/unit/test__opentelemetry_metrics.py b/packages/google-cloud-storage/tests/unit/test__opentelemetry_metrics.py index 275baa8f7c5c..79a2e76eaee5 100644 --- a/packages/google-cloud-storage/tests/unit/test__opentelemetry_metrics.py +++ b/packages/google-cloud-storage/tests/unit/test__opentelemetry_metrics.py @@ -1266,7 +1266,21 @@ class _NoneCodeErr(Exception): def code(self): return None + class _NonTupleArgsErr(Exception): + args = 123 # type: ignore[assignment] + assert _opentelemetry_metrics.classify_error_type(_NoneCodeErr()) == "UNKNOWN" + assert _opentelemetry_metrics.classify_error_type(_NonTupleArgsErr()) == "UNKNOWN" + assert ( + _opentelemetry_metrics.classify_error_type(concurrent.futures.CancelledError()) + == "CANCELLED" + ) + assert ( + _opentelemetry_metrics._is_subclass_by_module( + mock.MagicMock(__mro__=123), "google.cloud.storage", {"DataCorruption"} + ) + is False + ) assert ( _opentelemetry_metrics.classify_error_type( urllib3.exceptions.NewConnectionError(None, "conn err")