From 43d1e94caf36dce78162e9fe38a7c6c2fd1dd47f Mon Sep 17 00:00:00 2001 From: Shradha Katyal Date: Thu, 15 Oct 2026 15:44:24 +0000 Subject: [PATCH 1/5] 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/5] 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/5] 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") From 49daee8669a9e78d1e5f6162c47f90c00f75f317 Mon Sep 17 00:00:00 2001 From: Shradha Katyal Date: Sun, 25 Oct 2026 18:13:38 +0000 Subject: [PATCH 4/5] fix(storage): align core OTel metrics module with Monarch schema - 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] --- .../cloud/storage/_opentelemetry_metrics.py | 352 +++++++++++++++--- .../tests/unit/test__opentelemetry_metrics.py | 341 ++++++++++++++--- 2 files changed, 589 insertions(+), 104 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 27675bc298a0..f9583d5baba4 100644 --- a/packages/google-cloud-storage/google/cloud/storage/_opentelemetry_metrics.py +++ b/packages/google-cloud-storage/google/cloud/storage/_opentelemetry_metrics.py @@ -26,8 +26,18 @@ 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 dataclasses import dataclass, field +from typing import ( + Any, + Deque, + Dict, + FrozenSet, + Iterator, + List, + Optional, + Set, + Tuple, +) from urllib.parse import urlsplit from google.cloud.storage.version import __version__ @@ -60,16 +70,17 @@ 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_CLIENT_RETRIES = "gcp.storage.client.retries" 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_TTFB = "gcp.storage.client.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, + METRIC_STORAGE_CLIENT_TTFB, ) _SIZE_METRIC_NAMES = ( @@ -179,7 +190,6 @@ def is_debug_metrics_enabled( _COMMON_ATTRIBUTES: Dict[str, Any] = { "gcp.client.service": "storage", "gcp.client.version": __version__, - "gcp.client.repo": "googleapis/google-cloud-python", "gcp.client.artifact": "google-cloud-storage", } @@ -256,8 +266,9 @@ def _build_latency_buckets() -> Tuple[float, ...]: 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). + Uses 128 KiB increments up to 4 MiB, then doubles the increment on each + subsequent bucket while ``boundary <= 16 GiB`` (48 total buckets, matching + Go and Java SDKs). """ kb = 1024 mb = 1024 * kb @@ -269,10 +280,9 @@ def _build_size_buckets() -> Tuple[int, ...]: while boundary <= 16 * gb: boundaries.append(boundary) - if boundary < 4 * mb: - boundary += increment - else: - boundary *= 2 + boundary += increment + if boundary >= 4 * mb: + increment *= 2 return tuple(boundaries) @@ -362,12 +372,13 @@ def _iter_exception_chain(exc: BaseException) -> Iterator[BaseException]: visited.add(id(nested)) queue.append(nested) - 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 seq_attr in ("args", "errors"): + seq = getattr(current, seq_attr, ()) + if isinstance(seq, (tuple, list)): + for item in seq: + if isinstance(item, BaseException) and id(item) not in visited: + visited.add(id(item)) + queue.append(item) def _is_subclass_by_module( @@ -725,15 +736,96 @@ def normalize_url_template(url_or_path: Optional[str]) -> str: # --------------------------------------------------------------------------- -# 9. Logical Operation Context & Suppression (`ContextVar`) +# 9. Logical Operation Normalization, Context & Suppression (`ContextVar`) # --------------------------------------------------------------------------- -@dataclass(frozen=True) +_GRPC_LOGICAL_OPERATION_OVERRIDES: Dict[str, str] = { + "BidiReadObject": "ReadObject", + "BidiWriteObject": "WriteObject", +} + +_RPC_V2_METHODS: FrozenSet[str] = frozenset( + { + "BidiReadObject", + "BidiWriteObject", + "CancelResumableWrite", + "ComposeObject", + "CreateBucket", + "DeleteBucket", + "DeleteObject", + "GetBucket", + "GetIamPolicy", + "GetObject", + "ListBuckets", + "ListObjects", + "LockBucketRetentionPolicy", + "MoveObject", + "QueryWriteStatus", + "ReadObject", + "RestoreObject", + "RewriteObject", + "SetIamPolicy", + "StartResumableWrite", + "TestIamPermissions", + "UpdateBucket", + "UpdateObject", + "WriteObject", + } +) + + +def normalize_logical_operation(operation: Optional[str]) -> str: + """Normalizes an RPC path or method name into a short logical ``gcp.storage.operation``.""" + if not operation or not isinstance(operation, str): + return "UNKNOWN" + cleaned = operation.strip().strip("/") + if not cleaned: + return "UNKNOWN" + short_name = cleaned.rsplit("/", 1)[-1] if "/" in cleaned else cleaned + return _GRPC_LOGICAL_OPERATION_OVERRIDES.get(short_name, short_name) + + +def resolve_rpc_method( + operation: Optional[str], rpc_method: Optional[str] = None +) -> str: + """Resolves the full RPC path (e.g. ``google.storage.v2.Storage/ReadObject``) for ``rpc.method``.""" + raw = rpc_method if (rpc_method and isinstance(rpc_method, str)) else operation + if not raw or not isinstance(raw, str): + return "UNKNOWN" + cleaned = raw.strip().strip("/") + if not cleaned or cleaned == "UNKNOWN": + return "UNKNOWN" + if "/" in cleaned: + return cleaned + if cleaned in _RPC_V2_METHODS: + return f"google.storage.v2.Storage/{cleaned}" + if "." in cleaned: + return cleaned + return f"google.storage.v1.Storage/{cleaned}" + + +@dataclass class _OperationState: """Represents the active logical GCS Veneer operation in the current context.""" client_method: str + operation: str rpc_method: str suppressed: bool = False + start_time: float = field(default_factory=time.perf_counter) + finished: bool = False + last_failure_error_type: Optional[str] = None + recorded_attempts: int = 0 + + def mark_attempt_failed(self, error_type: str) -> None: + """Records the error type of a failed attempt for a subsequent retry.""" + if error_type and error_type != "OK": + self.last_failure_error_type = error_type + + def consume_retry_error(self) -> Optional[str]: + """Consumes and returns the prior failed attempt's error type, if any.""" + prior = self.last_failure_error_type + self.last_failure_error_type = None + return prior _OP_STATE: ContextVar[Optional[_OperationState]] = ContextVar( @@ -743,15 +835,19 @@ class _OperationState: def get_current_operation() -> Optional[_OperationState]: """Returns the current logical `_OperationState` in this execution context, if any.""" - return _OP_STATE.get() + op = _OP_STATE.get() + if op is not None and op.finished: + return None + return op @contextlib.contextmanager def start_operation( recorder: Optional["_MetricsRecorder"], client_method: str, - rpc_method: str, + operation: str, suppress: bool = False, + rpc_method: Optional[str] = None, ) -> Iterator[Optional[_OperationState]]: """Context manager tracking a logical user-facing GCS operation. @@ -763,24 +859,30 @@ def start_operation( 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"``). + operation: Short logical GCS operation name + (e.g. ``"ReadObject"``, ``"GetObject"``). 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). + rpc_method: Optional full RPC path + (e.g. ``"google.storage.v2.Storage/BidiReadObject"``). If omitted, + inferred from ``operation``. """ - existing = _OP_STATE.get() + existing = get_current_operation() if existing is not None: if suppress and not existing.suppressed: suppressed_child = _OperationState( client_method=existing.client_method, + operation=existing.operation, rpc_method=existing.rpc_method, suppressed=True, + start_time=existing.start_time, ) child_token = _OP_STATE.set(suppressed_child) try: yield suppressed_child finally: + suppressed_child.finished = True _OP_STATE.reset(child_token) else: yield existing @@ -793,13 +895,19 @@ def start_operation( yield None return + resolved_operation = normalize_logical_operation(operation) + resolved_rpc_method = resolve_rpc_method( + resolved_operation, rpc_method=rpc_method or operation + ) + start_time = time.perf_counter() state = _OperationState( client_method=client_method, - rpc_method=rpc_method, + operation=resolved_operation, + rpc_method=resolved_rpc_method, suppressed=suppress, + start_time=start_time, ) token = _OP_STATE.set(state) - start_time = time.perf_counter() error_type = "OK" try: yield state @@ -807,17 +915,66 @@ def start_operation( error_type = classify_error_type(exc) raise finally: + state.finished = True 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, + operation=resolved_operation, error_type=error_type, + rpc_method=resolved_rpc_method, ) +def record_pending_retry( + recorder: Optional["_MetricsRecorder"], + operation: Optional[str] = None, + network_path: Optional[str] = None, +) -> None: + """Emits a pending ``gcp.storage.client.retries`` increment if a prior attempt failed.""" + if recorder is None or not getattr(recorder, "standard_enabled", False): + return + current_op = get_current_operation() + if current_op is None: + return + prior_error = current_op.consume_retry_error() + if prior_error is not None: + recorder.record_retry( + operation=operation or current_op.operation, + error_type=prior_error, + network_path=network_path, + ) + + +def handle_unrecorded_retryable_failure( + recorder: Optional["_MetricsRecorder"], + exc: BaseException, + attempts_before: int, + will_retry: bool, + network_path: Optional[str] = None, +) -> None: + """Tracks retry state when a failed attempt bypassed transport-level attempt recording.""" + if recorder is None or not getattr(recorder, "standard_enabled", False): + return + current_op = get_current_operation() + if current_op is None: + return + if current_op.recorded_attempts > attempts_before: + return + current_op.recorded_attempts += 1 + prior_error = current_op.consume_retry_error() + if prior_error is not None: + recorder.record_retry( + operation=current_op.operation, + error_type=prior_error, + network_path=network_path, + ) + if will_retry: + current_op.mark_attempt_failed(classify_error_type(exc)) + + def get_active_recorder(obj: Any) -> Optional["_MetricsRecorder"]: """Returns ``obj._metrics_recorder`` if it is an active ``_MetricsRecorder``.""" recorder = getattr(obj, "_metrics_recorder", None) @@ -899,6 +1056,11 @@ def __init__( unit="{attempt}", description="Number of physical network transport attempts executed.", ) + self.retries_counter = self.meter.create_counter( + METRIC_STORAGE_CLIENT_RETRIES, + unit="{retry}", + description="Number of transport retries executed after a failed attempt.", + ) self.request_body_size = self.meter.create_histogram( METRIC_STORAGE_REQUEST_BODY_SIZE, unit="By", @@ -909,11 +1071,12 @@ def __init__( unit="By", description="Application payload bytes delivered to the caller.", ) - self.operation_ttfb = self.meter.create_histogram( - METRIC_STORAGE_OPERATION_TTFB, + self.ttfb = self.meter.create_histogram( + METRIC_STORAGE_CLIENT_TTFB, unit="s", description="Time elapsed from attempt initiation to receipt of the first response byte.", ) + self.operation_ttfb = self.ttfb self.errors_counter = self.meter.create_counter( METRIC_STORAGE_CLIENT_ERRORS, unit="{error}", @@ -925,8 +1088,10 @@ def __init__( self.client_request_duration = None self.operations_counter = None self.attempts_counter = None + self.retries_counter = None self.request_body_size = None self.response_body_size = None + self.ttfb = None self.operation_ttfb = None self.errors_counter = None @@ -943,13 +1108,13 @@ def _build_operation_attributes( """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) + current_op = get_current_operation() + raw_op = operation or (current_op.operation 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 raw_op is not None: + attrs["gcp.storage.operation"] = normalize_logical_operation(raw_op) if resolved_method is not None: attrs["gcp.client.method"] = resolved_method return attrs @@ -958,20 +1123,52 @@ def record_logical_operation( self, duration: float, client_method: str, - rpc_method: str, + operation: str, error_type: str = "OK", + rpc_method: Optional[str] = None, ) -> None: """Records end-to-end logical operation duration and count.""" if not self.standard_enabled or self.meter is None: return - attrs = { + logical_op = normalize_logical_operation(operation) + full_rpc_method = resolve_rpc_method( + logical_op, rpc_method=rpc_method or operation + ) + op_attrs = { **self._base_attributes, - "gcp.storage.operation": rpc_method, + "gcp.storage.operation": logical_op, "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) + duration_attrs = { + **op_attrs, + "rpc.method": full_rpc_method, + } + self.client_request_duration.record( + max(0.0, duration), attributes=duration_attrs + ) + self.operations_counter.add(1, attributes=op_attrs) + + def record_retry( + self, + operation: Optional[str] = None, + error_type: str = "UNKNOWN", + network_path: Optional[str] = None, + ) -> None: + """Records a transport retry attempt with the prior failed attempt's error type.""" + if not self.standard_enabled or self.meter is None: + return + current_op = get_current_operation() + raw_op = operation or (current_op.operation if current_op else None) + retry_attrs: Dict[str, Any] = { + **self._base_attributes, + "error.type": error_type, + } + if raw_op is not None: + retry_attrs["gcp.storage.operation"] = normalize_logical_operation(raw_op) + if network_path is not None: + retry_attrs["network.path"] = network_path + self.retries_counter.add(1, attributes=retry_attrs) def record_http_attempt( self, @@ -987,11 +1184,24 @@ def record_http_attempt( """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 + del client_method + current_op = get_current_operation() + raw_op = operation or (current_op.operation if current_op else None) + resolved_op = ( + normalize_logical_operation(raw_op) if raw_op is not None else None ) + if current_op is not None: + current_op.recorded_attempts += 1 + prior_error = current_op.consume_retry_error() + if prior_error is not None: + self.record_retry( + operation=resolved_op, + error_type=prior_error, + network_path=network_path, + ) + if error_type != "OK": + current_op.mark_attempt_failed(error_type) + valid_code = ( isinstance(status_code, int) and not isinstance(status_code, bool) @@ -1033,29 +1243,47 @@ def record_http_attempt( } 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 + if network_path is not None: + error_attrs["network.path"] = network_path self.errors_counter.add(1, attributes=error_attrs) def record_rpc_attempt( self, duration: float, - operation: str, + operation: Optional[str] = None, status_code: Optional[str] = None, error_type: str = "OK", - client_method: Optional[str] = None, + rpc_method: Optional[str] = None, network_path: Optional[str] = None, + op_state: Optional[_OperationState] = 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 + active_op = op_state if op_state is not None else get_current_operation() + raw_rpc = ( + rpc_method or operation or (active_op.rpc_method if active_op else None) ) + resolved_rpc_method = resolve_rpc_method(operation, rpc_method=raw_rpc) + raw_op = operation or rpc_method or (active_op.operation if active_op else None) + resolved_op = normalize_logical_operation(raw_op) + + if active_op is not None: + active_op.recorded_attempts += 1 + prior_error = active_op.consume_retry_error() + if prior_error is not None: + self.record_retry( + operation=resolved_op, + error_type=prior_error, + network_path=network_path, + ) + if error_type != "OK": + active_op.mark_attempt_failed(error_type) # 1. rpc.client.call.duration attributes rpc_attrs: Dict[str, Any] = { - **op_attrs, + **self._base_attributes, + "rpc.method": resolved_rpc_method, "error.type": error_type, } if status_code is not None: @@ -1066,7 +1294,7 @@ def record_rpc_attempt( # 2. gcp.storage.client.attempts attributes attempt_attrs: Dict[str, Any] = { **self._base_attributes, - "gcp.storage.operation": operation, + "gcp.storage.operation": resolved_op, "error.type": error_type, } if status_code is not None: @@ -1080,9 +1308,12 @@ def record_rpc_attempt( # 3. gcp.storage.client.errors attributes if error_type != "OK": error_attrs: Dict[str, Any] = { - **op_attrs, + **self._base_attributes, + "gcp.storage.operation": resolved_op, "error.type": error_type, } + if network_path is not None: + error_attrs["network.path"] = network_path self.errors_counter.add(1, attributes=error_attrs) def record_request_body_size( @@ -1096,7 +1327,7 @@ def record_request_body_size( return if not isinstance(size, int) or isinstance(size, bool): return - current_op = _OP_STATE.get() + current_op = get_current_operation() if current_op is not None and current_op.suppressed: return attrs = self._build_operation_attributes( @@ -1115,7 +1346,7 @@ def record_response_body_size( return if not isinstance(size, int) or isinstance(size, bool): return - current_op = _OP_STATE.get() + current_op = get_current_operation() if current_op is not None and current_op.suppressed: return attrs = self._build_operation_attributes( @@ -1127,18 +1358,21 @@ def record_ttfb( self, duration: float, operation: Optional[str] = None, - client_method: Optional[str] = None, + network_path: 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() + current_op = get_current_operation() 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) + raw_op = operation or (current_op.operation if current_op else None) + attrs: Dict[str, Any] = self._base_attributes.copy() + if raw_op is not None: + attrs["gcp.storage.operation"] = normalize_logical_operation(raw_op) + if network_path is not None: + attrs["network.path"] = network_path + self.ttfb.record(max(0.0, duration), attributes=attrs) def shutdown(self) -> None: """Flushes pending metrics on a custom MeterProvider if configured.""" 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 79a2e76eaee5..f96cb39a7ca5 100644 --- a/packages/google-cloud-storage/tests/unit/test__opentelemetry_metrics.py +++ b/packages/google-cloud-storage/tests/unit/test__opentelemetry_metrics.py @@ -230,9 +230,14 @@ def test_otel_missing_disables_metrics(monkeypatch): def test_get_common_attributes(): """Verify common attributes conform to GCS OTel specification.""" attrs = _opentelemetry_metrics.get_common_attributes() + assert set(attrs.keys()) == { + "gcp.client.service", + "gcp.client.version", + "gcp.client.artifact", + } + assert "gcp.client.repo" not in attrs assert attrs["gcp.client.service"] == "storage" assert attrs["gcp.client.version"] == __version__ - assert attrs["gcp.client.repo"] == "googleapis/google-cloud-python" assert attrs["gcp.client.artifact"] == "google-cloud-storage" attrs["gcp.client.service"] = "mutated" @@ -316,12 +321,13 @@ def test_histogram_bucket_boundaries(): assert list(latency) == sorted(latency) sizes = _opentelemetry_metrics.SIZE_HISTOGRAM_BOUNDARIES - assert len(sizes) == 45 + assert len(sizes) == 48 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 sizes[33] == int(4.25 * 1024 * 1024) + assert sizes[34] == int(4.75 * 1024 * 1024) + assert sizes[-1] == 8593866752 assert list(sizes) == sorted(sizes) @@ -1111,7 +1117,7 @@ def test_metrics_recorder_recording_and_shutdown(monkeypatch): "By", "Application payload bytes delivered to the caller.", ), - "gcp.storage.client.operation.ttfb": ( + "gcp.storage.client.ttfb": ( "s", "Time elapsed from attempt initiation to receipt of the first response byte.", ), @@ -1129,6 +1135,10 @@ def test_metrics_recorder_recording_and_shutdown(monkeypatch): "{attempt}", "Number of physical network transport attempts executed.", ), + "gcp.storage.client.retries": ( + "{retry}", + "Number of transport retries executed after a failed attempt.", + ), "gcp.storage.client.errors": ( "{error}", "Number of physical network transport attempts terminating with a non-OK status.", @@ -1137,7 +1147,10 @@ def test_metrics_recorder_recording_and_shutdown(monkeypatch): # Record within a logical operation context with _opentelemetry_metrics.start_operation( - recorder, "storage.Blob.upload_from_string", "storage.objects.insert" + recorder, + "storage.Blob.upload_from_string", + "WriteObject", + rpc_method="google.storage.v2.Storage/WriteObject", ): recorder.record_http_attempt( duration=0.05, @@ -1153,10 +1166,30 @@ def test_metrics_recorder_recording_and_shutdown(monkeypatch): url_template="/upload/storage/v1/b/{bucket}/o", status_code=200, error_type="OK", + network_path="cloudpath", ) recorder.record_request_body_size(1024) recorder.record_response_body_size(512) - recorder.record_ttfb(0.01) + recorder.record_ttfb(0.01, network_path="cloudpath") + + # Logical operation metrics: duration has rpc.method; operations counter does not + recorder.client_request_duration.record.assert_called_once() + _, req_dur_kwargs = recorder.client_request_duration.record.call_args + assert req_dur_kwargs["attributes"]["gcp.storage.operation"] == "WriteObject" + assert ( + req_dur_kwargs["attributes"]["rpc.method"] + == "google.storage.v2.Storage/WriteObject" + ) + assert ( + req_dur_kwargs["attributes"]["gcp.client.method"] + == "storage.Blob.upload_from_string" + ) + assert req_dur_kwargs["attributes"]["error.type"] == "OK" + + recorder.operations_counter.add.assert_called_once() + _, op_cnt_kwargs = recorder.operations_counter.add.call_args + assert op_cnt_kwargs["attributes"]["gcp.storage.operation"] == "WriteObject" + assert "rpc.method" not in op_cnt_kwargs["attributes"] assert recorder.http_request_duration.record.call_count == 2 _, http_dur_kwargs = recorder.http_request_duration.record.call_args_list[0] @@ -1173,79 +1206,101 @@ def test_metrics_recorder_recording_and_shutdown(monkeypatch): 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"]["gcp.storage.operation"] == "WriteObject" 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"] + assert "gcp.client.method" not in attempt_kwargs["attributes"] - # Only the 503 attempt records an error + # Retry counter incremented once for the second attempt after 503 UNAVAILABLE + recorder.retries_counter.add.assert_called_once() + _, http_retry_kwargs = recorder.retries_counter.add.call_args + assert http_retry_kwargs["attributes"]["gcp.storage.operation"] == "WriteObject" + assert http_retry_kwargs["attributes"]["error.type"] == "UNAVAILABLE" + assert http_retry_kwargs["attributes"]["network.path"] == "cloudpath" + assert "gcp.client.method" not in http_retry_kwargs["attributes"] + + # Only the 503 attempt records an error (no gcp.client.method on errors) 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 err_kwargs["attributes"]["gcp.storage.operation"] == "WriteObject" + assert err_kwargs["attributes"]["network.path"] == "cloudpath" + assert "gcp.client.method" not in err_kwargs["attributes"] 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() + recorder.ttfb.record.assert_called_once() + _, ttfb_kwargs = recorder.ttfb.record.call_args + assert ttfb_kwargs["attributes"]["gcp.storage.operation"] == "WriteObject" + assert ttfb_kwargs["attributes"]["network.path"] == "cloudpath" + assert "gcp.client.method" not in ttfb_kwargs["attributes"] - # gRPC attempt recording (OK and non-OK) + # gRPC attempt recording (UNAVAILABLE -> OK retry sequence) 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.asyncio.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.asyncio.AsyncMultiRangeDownloader.download_ranges", - ) + recorder.retries_counter.add.reset_mock() + with _opentelemetry_metrics.start_operation( + recorder, + "storage.asyncio.AsyncMultiRangeDownloader.download_ranges", + "ReadObject", + rpc_method="google.storage.v2.Storage/BidiReadObject", + ): + recorder.record_rpc_attempt( + duration=0.03, + operation="ReadObject", + status_code="UNAVAILABLE", + error_type="UNAVAILABLE", + rpc_method="google.storage.v2.Storage/BidiReadObject", + network_path="directpath", + ) + recorder.record_rpc_attempt( + duration=0.02, + operation="ReadObject", + status_code="OK", + error_type="OK", + rpc_method="google.storage.v2.Storage/BidiReadObject", + network_path="directpath", + ) assert recorder.rpc_call_duration.record.call_count == 2 - _, rpc_dur_kwargs = recorder.rpc_call_duration.record.call_args_list[0] + _, rpc_dur_kwargs = recorder.rpc_call_duration.record.call_args_list[1] assert ( - rpc_dur_kwargs["attributes"]["gcp.storage.operation"] + rpc_dur_kwargs["attributes"]["rpc.method"] == "google.storage.v2.Storage/BidiReadObject" ) - assert ( - rpc_dur_kwargs["attributes"]["gcp.client.method"] - == "storage.asyncio.AsyncMultiRangeDownloader.download_ranges" - ) assert rpc_dur_kwargs["attributes"]["rpc.response.status_code"] == "OK" + assert rpc_dur_kwargs["attributes"]["error.type"] == "OK" assert rpc_dur_kwargs["attributes"]["network.path"] == "directpath" + assert "gcp.storage.operation" not in rpc_dur_kwargs["attributes"] + assert "gcp.client.method" not in rpc_dur_kwargs["attributes"] 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" - ) + _, rpc_attempt_kwargs = recorder.attempts_counter.add.call_args_list[1] + assert rpc_attempt_kwargs["attributes"]["gcp.storage.operation"] == "ReadObject" 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"] + assert "rpc.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"]["gcp.storage.operation"] == "ReadObject" assert rpc_err_kwargs["attributes"]["error.type"] == "UNAVAILABLE" + assert rpc_err_kwargs["attributes"]["network.path"] == "directpath" + assert "gcp.client.method" not in rpc_err_kwargs["attributes"] assert "rpc.response.status_code" not in rpc_err_kwargs["attributes"] + recorder.retries_counter.add.assert_called_once() + _, rpc_retry_kwargs = recorder.retries_counter.add.call_args + assert rpc_retry_kwargs["attributes"]["gcp.storage.operation"] == "ReadObject" + assert rpc_retry_kwargs["attributes"]["error.type"] == "UNAVAILABLE" + assert rpc_retry_kwargs["attributes"]["network.path"] == "directpath" + # Record outside of start_operation with status_code=None and callable .code() returning None recorder.record_http_attempt( duration=0.01, @@ -1334,9 +1389,205 @@ def test_metrics_recorder_disabled_no_ops(monkeypatch): # All recording methods are safe no-ops when disabled recorder.record_logical_operation(0.1, "m", "op") + recorder.record_retry("ReadObject", "UNAVAILABLE", "cloudpath") 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() + + +def test_normalize_logical_operation_and_resolve_rpc_method(): + assert _opentelemetry_metrics.normalize_logical_operation(None) == "UNKNOWN" + assert _opentelemetry_metrics.normalize_logical_operation("") == "UNKNOWN" + assert _opentelemetry_metrics.normalize_logical_operation(123) == "UNKNOWN" # type: ignore[arg-type] + assert _opentelemetry_metrics.normalize_logical_operation("///") == "UNKNOWN" + assert ( + _opentelemetry_metrics.normalize_logical_operation( + "google.storage.v2.Storage/BidiReadObject" + ) + == "ReadObject" + ) + assert ( + _opentelemetry_metrics.normalize_logical_operation( + "/google.storage.v2.Storage/BidiWriteObject" + ) + == "WriteObject" + ) + assert ( + _opentelemetry_metrics.normalize_logical_operation( + "google.storage.v2.Storage/GetObject" + ) + == "GetObject" + ) + assert ( + _opentelemetry_metrics.normalize_logical_operation("GetNotificationConfig") + == "GetNotificationConfig" + ) + + assert _opentelemetry_metrics.resolve_rpc_method(None, None) == "UNKNOWN" + assert _opentelemetry_metrics.resolve_rpc_method("", "") == "UNKNOWN" + assert _opentelemetry_metrics.resolve_rpc_method("///", None) == "UNKNOWN" + assert _opentelemetry_metrics.resolve_rpc_method("UNKNOWN", None) == "UNKNOWN" + assert ( + _opentelemetry_metrics.resolve_rpc_method( + "ReadObject", "google.storage.v2.Storage/BidiReadObject" + ) + == "google.storage.v2.Storage/BidiReadObject" + ) + assert ( + _opentelemetry_metrics.resolve_rpc_method("ReadObject") + == "google.storage.v2.Storage/ReadObject" + ) + assert ( + _opentelemetry_metrics.resolve_rpc_method("GetNotificationConfig") + == "google.storage.v1.Storage/GetNotificationConfig" + ) + assert ( + _opentelemetry_metrics.resolve_rpc_method("storage.objects.get") + == "storage.objects.get" + ) + + +def test_iter_exception_chain_traverses_errors_list(): + wrapped = api_exceptions.Aborted( + "aborted", + errors=[socket.timeout("nested timeout in errors")], + ) + chain = list(_opentelemetry_metrics._iter_exception_chain(wrapped)) + assert any(isinstance(item, socket.timeout) for item in chain) + + class _NonSeqErrorsErr(Exception): + errors = 123 + + assert _opentelemetry_metrics.classify_error_type(_NonSeqErrorsErr()) == "UNKNOWN" + + +def test_operation_state_and_retry_helpers(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" + ) + disabled_recorder = _opentelemetry_metrics._MetricsRecorder(enable_metrics=False) + + # No-ops when recorder is None, disabled, or no active operation + _opentelemetry_metrics.record_pending_retry(None) + _opentelemetry_metrics.record_pending_retry(disabled_recorder) + _opentelemetry_metrics.record_pending_retry(recorder) + _opentelemetry_metrics.handle_unrecorded_retryable_failure( + None, RuntimeError("e"), attempts_before=0, will_retry=True + ) + _opentelemetry_metrics.handle_unrecorded_retryable_failure( + disabled_recorder, RuntimeError("e"), attempts_before=0, will_retry=True + ) + _opentelemetry_metrics.handle_unrecorded_retryable_failure( + recorder, RuntimeError("e"), attempts_before=0, will_retry=True + ) + + # Standalone record_retry and record_ttfb outside start_operation without optional attrs + recorder.record_retry(operation=None, error_type="UNAVAILABLE", network_path=None) + _, retry_no_op_kw = recorder.retries_counter.add.call_args + assert "gcp.storage.operation" not in retry_no_op_kw["attributes"] + assert "network.path" not in retry_no_op_kw["attributes"] + + recorder.record_ttfb(0.02, operation=None, network_path=None) + _, ttfb_no_op_kw = recorder.ttfb.record.call_args + assert "gcp.storage.operation" not in ttfb_no_op_kw["attributes"] + assert "network.path" not in ttfb_no_op_kw["attributes"] + + recorder.retries_counter.add.reset_mock() + with _opentelemetry_metrics.start_operation( + recorder, + "storage.asyncio.AsyncMultiRangeDownloader.open", + "ReadObject", + rpc_method="google.storage.v2.Storage/BidiReadObject", + ) as op_state: + assert op_state is not None + op_state.mark_attempt_failed("OK") + op_state.mark_attempt_failed("") + assert op_state.last_failure_error_type is None + + # Pending retry when no failure occurred is a no-op + _opentelemetry_metrics.record_pending_retry(recorder) + recorder.retries_counter.add.assert_not_called() + + # When transport already recorded the attempt, handle_unrecorded_retryable_failure skips + op_state.recorded_attempts = 1 + _opentelemetry_metrics.handle_unrecorded_retryable_failure( + recorder, + api_exceptions.ServiceUnavailable("503"), + attempts_before=0, + will_retry=True, + network_path="directpath", + ) + assert op_state.recorded_attempts == 1 + assert op_state.last_failure_error_type is None + + # Unrecorded failure 1 (will_retry=True) -> marks failure, no retry yet + _opentelemetry_metrics.handle_unrecorded_retryable_failure( + recorder, + api_exceptions.ServiceUnavailable("503"), + attempts_before=1, + will_retry=True, + network_path="unknown", + ) + assert op_state.recorded_attempts == 2 + assert op_state.last_failure_error_type == "UNAVAILABLE" + recorder.retries_counter.add.assert_not_called() + + # Consecutive unrecorded failure 2 (will_retry=True) -> emits retry for failure 1, marks failure 2 + _opentelemetry_metrics.handle_unrecorded_retryable_failure( + recorder, + api_exceptions.DeadlineExceeded("timeout"), + attempts_before=2, + will_retry=True, + network_path="cloudpath", + ) + assert op_state.recorded_attempts == 3 + assert op_state.last_failure_error_type == "DEADLINE_EXCEEDED" + recorder.retries_counter.add.assert_called_once() + assert ( + recorder.retries_counter.add.call_args.kwargs["attributes"]["error.type"] + == "UNAVAILABLE" + ) + + # Unrecorded terminal failure 3 (will_retry=False) -> emits retry for failure 2, leaves error cleared + _opentelemetry_metrics.handle_unrecorded_retryable_failure( + recorder, + api_exceptions.PermissionDenied("403"), + attempts_before=3, + will_retry=False, + network_path="cloudpath", + ) + assert op_state.recorded_attempts == 4 + assert op_state.last_failure_error_type is None + assert recorder.retries_counter.add.call_count == 2 + assert ( + recorder.retries_counter.add.call_args.kwargs["attributes"]["error.type"] + == "DEADLINE_EXCEEDED" + ) + + # Mark failure and consume via record_pending_retry + op_state.mark_attempt_failed("ABORTED") + _opentelemetry_metrics.record_pending_retry(recorder, network_path="directpath") + assert recorder.retries_counter.add.call_count == 3 + _, pending_kw = recorder.retries_counter.add.call_args + assert pending_kw["attributes"]["gcp.storage.operation"] == "ReadObject" + assert pending_kw["attributes"]["error.type"] == "ABORTED" + assert pending_kw["attributes"]["network.path"] == "directpath" + + import contextvars + + copied_ctx = contextvars.copy_context() + + # Once start_operation exits, state.finished is True even in copied contexts + assert op_state.finished is True + assert copied_ctx.run(_opentelemetry_metrics.get_current_operation) is None From cdea802ab7b5d128ef7227f5bd81598c5cabc3a0 Mon Sep 17 00:00:00 2001 From: Shradha Katyal Date: Mon, 26 Oct 2026 09:57:02 +0000 Subject: [PATCH 5/5] fix(storage): scope OTel histogram views, harden URI parsing, and unify 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] --- .../cloud/storage/_opentelemetry_metrics.py | 49 ++++++++++++++----- .../tests/unit/test__opentelemetry_metrics.py | 30 ++++++++---- 2 files changed, 57 insertions(+), 22 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 f9583d5baba4..f511eff47eea 100644 --- a/packages/google-cloud-storage/google/cloud/storage/_opentelemetry_metrics.py +++ b/packages/google-cloud-storage/google/cloud/storage/_opentelemetry_metrics.py @@ -187,6 +187,8 @@ def is_debug_metrics_enabled( # --------------------------------------------------------------------------- # 5. Standard Common Attributes & Meter Provider # --------------------------------------------------------------------------- +_METER_NAME = "google.cloud.storage" + _COMMON_ATTRIBUTES: Dict[str, Any] = { "gcp.client.service": "storage", "gcp.client.version": __version__, @@ -227,7 +229,7 @@ def get_meter( ): return None return metrics.get_meter( - "google.cloud.storage", + _METER_NAME, __version__, meter_provider=meter_provider, ) @@ -309,6 +311,7 @@ def get_histogram_views() -> List[Any]: views.append( View( instrument_name=name, + meter_name=_METER_NAME, aggregation=ExplicitBucketHistogramAggregation( boundaries=LATENCY_HISTOGRAM_BOUNDARIES ), @@ -318,6 +321,7 @@ def get_histogram_views() -> List[Any]: views.append( View( instrument_name=name, + meter_name=_METER_NAME, aggregation=ExplicitBucketHistogramAggregation( boundaries=SIZE_HISTOGRAM_BOUNDARIES ), @@ -507,10 +511,8 @@ def classify_error_type( {"NameResolutionError"}, ): return "DNS_FAILURE" - if isinstance(item, ConnectionRefusedError): - return "CONNECTION_REFUSED" - if isinstance(item, ConnectionResetError): - return "CONNECTION_RESET" + if isinstance(item, (ConnectionRefusedError, ConnectionResetError)): + return "CONNECTION_ERROR" # 8. Timeouts (Note: urllib3.exceptions.NewConnectionError subclasses ConnectTimeoutError # even for non-timeout connection failures, so exclude NewConnectionError here.) @@ -537,7 +539,7 @@ def classify_error_type( ): return "TIMEOUT" - # 9. Generic Connectivity Failure + # 9. Generic Connection / Network Transport Failure for item in chain: if ( _is_subclass_by_module( @@ -552,7 +554,7 @@ def classify_error_type( ) or isinstance(item, (OSError, http.client.HTTPException)) ): - return "CONNECTIVITY" + return "CONNECTION_ERROR" # 10. Explicit HTTP status_code fallback if valid_status and status_code is not None: @@ -708,7 +710,10 @@ def normalize_url_template(url_or_path: Optional[str]) -> str: if not url_or_path or not isinstance(url_or_path, str): return "UNKNOWN" - parsed = urlsplit(url_or_path) + try: + parsed = urlsplit(url_or_path) + except ValueError: + return "UNKNOWN" path = parsed.path or "" query = parsed.query or "" @@ -993,12 +998,30 @@ def _extract_server_address(api_endpoint: Optional[str]) -> str: endpoint = api_endpoint.strip() if not endpoint: return DEFAULT_SERVER_ADDRESS - if ":///" in endpoint: - endpoint = "https://" + endpoint.split(":///", 1)[1].lstrip("/") - elif "://" not in endpoint: + for grpc_prefix in ( + "dns:", + "google-c2p:", + "google-c2p-experimental:", + "xds:", + "ipv4:", + "ipv6:", + ): + if endpoint.startswith(grpc_prefix): + remainder = endpoint[len(grpc_prefix) :] + if remainder.startswith("//"): + remainder = remainder[2:] + remainder = ( + remainder.split("/", 1)[1] if "/" in remainder else remainder + ) + endpoint = remainder.lstrip("/") + break + if "://" not in endpoint: endpoint = "https://" + endpoint.lstrip("/") - parsed = urlsplit(endpoint) - return parsed.hostname or DEFAULT_SERVER_ADDRESS + try: + parsed = urlsplit(endpoint) + return parsed.hostname or DEFAULT_SERVER_ADDRESS + except ValueError: + return DEFAULT_SERVER_ADDRESS class _MetricsRecorder: 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 f96cb39a7ca5..2e10d433cbe5 100644 --- a/packages/google-cloud-storage/tests/unit/test__opentelemetry_metrics.py +++ b/packages/google-cloud-storage/tests/unit/test__opentelemetry_metrics.py @@ -312,6 +312,7 @@ def test_get_meter(monkeypatch): def test_histogram_bucket_boundaries(): """Verify custom latency and byte-size bucket boundaries match specification.""" latency = _opentelemetry_metrics.LATENCY_HISTOGRAM_BOUNDARIES + assert len(latency) == 165 assert latency[0] == 0.0 assert latency[1] == 0.002 assert latency[49] == 0.098 @@ -340,9 +341,10 @@ def __init__(self, boundaries): self.boundaries = boundaries class FakeView: - def __init__(self, instrument_name, aggregation): + def __init__(self, instrument_name, aggregation, meter_name=None): self.instrument_name = instrument_name self.aggregation = aggregation + self.meter_name = meter_name fake_view_mod.ExplicitBucketHistogramAggregation = FakeAggregation fake_view_mod.View = FakeView @@ -361,6 +363,7 @@ def __init__(self, instrument_name, aggregation): views = _opentelemetry_metrics.get_histogram_views() assert len(views) == 6 + assert all(v.meter_name == "google.cloud.storage" for v in views) latency_views = [ v for v in views @@ -611,37 +614,37 @@ def code(self): wrapped_refused = requests.exceptions.ConnectionError(refused_err) assert ( _opentelemetry_metrics.classify_error_type(wrapped_refused) - == "CONNECTION_REFUSED" + == "CONNECTION_ERROR" ) 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" + _opentelemetry_metrics.classify_error_type(wrapped_reset) == "CONNECTION_ERROR" ) - # 10. Generic Connectivity + # 10. Generic Connection Errors assert ( _opentelemetry_metrics.classify_error_type( requests.exceptions.ConnectionError("connection aborted") ) - == "CONNECTIVITY" + == "CONNECTION_ERROR" ) assert ( _opentelemetry_metrics.classify_error_type( requests.exceptions.ChunkedEncodingError("incomplete read") ) - == "CONNECTIVITY" + == "CONNECTION_ERROR" ) assert ( _opentelemetry_metrics.classify_error_type(OSError("network unreachable")) - == "CONNECTIVITY" + == "CONNECTION_ERROR" ) assert ( _opentelemetry_metrics.classify_error_type( http.client.IncompleteRead(b"partial", 10) ) - == "CONNECTIVITY" + == "CONNECTION_ERROR" ) # 11. Explicit status_code fallback when exception is generic @@ -875,6 +878,7 @@ def code(self): ("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"), + ("https://[::1/storage/v1/b", "UNKNOWN"), ], ) def test_normalize_url_template(raw_url, expected): @@ -1061,6 +1065,14 @@ async def run_all(): ("https://[::1]:8080", "::1"), ("http://localhost:9023/storage/v1", "localhost"), ("https://", "storage.googleapis.com"), + ("https://[::1", "storage.googleapis.com"), + ("[::1", "storage.googleapis.com"), + ("dns://8.8.8.8/custom.endpoint.com:443", "custom.endpoint.com"), + ("dns:custom.endpoint.com:443", "custom.endpoint.com"), + ("dns://custom.endpoint.com:443", "custom.endpoint.com"), + ("dns://8.8.8.8/", "storage.googleapis.com"), + ("ipv4:127.0.0.1:443", "127.0.0.1"), + ("ipv6:[::1]:443", "::1"), ], ) def test_extract_server_address(endpoint, expected): @@ -1340,7 +1352,7 @@ class _NonTupleArgsErr(Exception): _opentelemetry_metrics.classify_error_type( urllib3.exceptions.NewConnectionError(None, "conn err") ) - == "CONNECTIVITY" + == "CONNECTION_ERROR" ) # Shutdown flushes custom meter_provider and swallows force_flush errors gracefully