From 95351467dbb9a540f316529af618c2b61395f05c Mon Sep 17 00:00:00 2001 From: chalmer lowe Date: Wed, 28 Oct 2026 12:20:21 -0400 Subject: [PATCH 1/4] feat(api_core): implement L2/L3 method error logging and L4 transport attempt logging --- .../google/api_core/_observability.py | 542 ++++++++++++++++-- .../google/api_core/client_logging.py | 27 +- .../google/api_core/gapic_v1/method.py | 134 ++++- .../tests/asyncio/gapic/test_method_async.py | 60 ++ .../tests/unit/gapic/test_method.py | 81 +++ .../tests/unit/test_client_logging.py | 18 + .../tests/unit/test_observability.py | 405 +++++++++++++ 7 files changed, 1184 insertions(+), 83 deletions(-) diff --git a/packages/google-api-core/google/api_core/_observability.py b/packages/google-api-core/google/api_core/_observability.py index e487687f5673..ef1d4faa840d 100644 --- a/packages/google-api-core/google/api_core/_observability.py +++ b/packages/google-api-core/google/api_core/_observability.py @@ -20,11 +20,12 @@ import contextlib import functools +import logging import traceback import urllib.parse from typing import TYPE_CHECKING, Any, Callable, Sequence -from google.api_core import _feature_gating_helpers, exceptions +from google.api_core import _feature_gating_helpers, client_logging, exceptions from google.api_core.client_options import ClientOptions try: @@ -288,6 +289,231 @@ def build_telemetry_error_attributes( return attrs +_LOGGER = logging.getLogger(__name__) + + +def _decode_method_name(method: str | bytes | None) -> str: + """Normalizes a string or bytes RPC method name to a UTF-8 string. + + Args: + method (Optional[Union[str, bytes]]): Raw RPC method name from call details. + + Returns: + str: Decoded string method name (or empty string if None). + """ + if isinstance(method, bytes): + return method.decode("utf-8", errors="replace") + return str(method or "") + + +def extract_rpc_identity( + method_name: str | bytes | None, +) -> tuple[str, str, str]: + """Extracts (full_rpc_name, service_name, rpc_method_name) from a gRPC method path. + + Args: + method_name (Optional[Union[str, bytes]]): Explicit gRPC method path + (e.g. "/google.cloud.secretmanager.v1.SecretManagerService/GetSecret"). + + Returns: + tuple[str, str, str]: A 3-tuple of (full_rpc_name, service_name, rpc_method_name). + """ + method_str = _decode_method_name(method_name).lstrip("/") + service, _, method = method_str.rpartition("/") + return method_str, service, method + + +def _resolve_rpc_system(kind: str | None = "grpc") -> str: + """Normalizes a transport kind ('grpc', 'rest', 'rest_asyncio', 'http') to an OTel rpc.system.name. + + Args: + kind (Optional[str]): The transport kind identifier. + + Returns: + str: "http" for REST/HTTP transports, or "grpc" otherwise. + """ + if isinstance(kind, str) and kind.startswith(("rest", "http")): + return "http" + return "grpc" + + +def _get_service_logger( + method_name: str | bytes | None = None, + target: Any = None, +) -> logging.Logger: + """Resolves a service-scoped logger matching GOOGLE_SDK_PYTHON_LOGGING_SCOPE. + + 1. If `target` has a `__module__` starting with `google.` (excluding `google.api_core`), + returns `logging.getLogger(target.__module__)`. + 2. Otherwise, if `method_name` is a full gRPC method path (e.g. + `/google.cloud.secretmanager.v1.SecretManagerService/GetSecret`), extracts + the service package prefix (`google.cloud.secretmanager.v1`). + 3. Falls back to `logging.getLogger("google.api_core._observability")`. + + Args: + method_name (Optional[Union[str, bytes]]): Full gRPC method path. + target (Optional[Any]): The low-level RPC target callable. + + Returns: + logging.Logger: The resolved service-scoped logger. + """ + target_mod = getattr(target, "__module__", None) + if ( + isinstance(target_mod, str) + and target_mod.startswith("google.") + and not target_mod.startswith("google.api_core") + ): + return logging.getLogger(target_mod) + if method_name: + # Example transformation: + # Input: "/google.cloud.secretmanager.v1.SecretManagerService/GetSecret" + # service_name: "google.cloud.secretmanager.v1.SecretManagerService" + # pkg: "google.cloud.secretmanager.v1" + _, service_name, _ = extract_rpc_identity(method_name) + pkg = service_name.rpartition(".")[0] + if pkg.startswith("google."): + return logging.getLogger(pkg) + return _LOGGER + + +def _is_failed_attempt( + norm_status: str | None, + exc: BaseException | None, +) -> bool: + """Determines whether a transport attempt ended with an error or exception. + + Args: + norm_status (Optional[str]): Normalized status code string (e.g. "OK", "200", "503"). + exc (Optional[BaseException]): Exception raised or returned during the attempt. + + Returns: + bool: True if the attempt failed, False if it succeeded. + """ + if exc is not None: + return True + if norm_status is None or norm_status == "OK": + return False + return not (norm_status.isdigit() and int(norm_status) < 400) + + +def _build_attempt_log_extra( + span: Any, + target: str | None, + status_code: int | str | None, + norm_status: str | None, + kind: str, + http_method: str | None, + exc: BaseException | None, + status_message: str | None, +) -> tuple[dict[str, Any], str | None]: + """Builds the structured extra dictionary and resolved status for a failed wire attempt. + + Args: + span (Any): Active OpenTelemetry span, if any. + target (Optional[str]): Full RPC method name (gRPC) or request URL (HTTP). + status_code (Optional[Union[int, str]]): Raw HTTP status code or gRPC status name. + norm_status (Optional[str]): Stringified status code if provided. + kind (str): Transport kind ("grpc", "rest", "rest_asyncio", or "http"). + http_method (Optional[str]): HTTP request method (e.g. "GET", "POST"). + exc (Optional[BaseException]): Exception raised or returned during the attempt. + status_message (Optional[str]): Explicit status message fallback. + + Returns: + tuple[dict[str, Any], Optional[str]]: A 2-tuple of (extra_payload, resolved_status). + """ + rpc_system = _resolve_rpc_system(kind) + error_attrs = build_telemetry_error_attributes( + exc, + status_code=norm_status, + status_message=status_message, + include_exception=True, + ) + resolved_status = norm_status or error_attrs.get("rpc.response.status_code") + extra_payload: dict[str, Any] = { + "rpc.system.name": rpc_system, + **error_attrs, + "rpc.response.status_code": resolved_status, + **extract_trace_context(span), + } + if rpc_system == "http": + extra_payload["url.full"] = target + extra_payload["httpRequest"] = { + "requestMethod": http_method, + "requestUrl": target, + "status": ( + int(status_code) + if isinstance(status_code, int) + or (isinstance(status_code, str) and status_code.isdigit()) + else status_code + ), + } + else: + extra_payload["rpc.method"] = target + + extra_payload["json_fields"] = dict(extra_payload) + return extra_payload, resolved_status + + +def log_transport_attempt( + span: Any = None, + target: str | None = None, + status_code: int | str | None = None, + kind: str = "grpc", + http_method: str | None = None, + exc: BaseException | None = None, + client_options: Any = None, + logger: logging.Logger | None = None, + status_message: str | None = None, +) -> None: + """Emits a DEBUG log for a failed wire attempt (retry/redirect/failure). + + Successful transport attempts (2xx HTTP or OK gRPC) return immediately without + emitting logs to avoid hot-path allocations. + + Args: + span (Optional[Any]): Active OpenTelemetry span, if any. + target (Optional[str]): Full RPC method name (gRPC) or request URL (HTTP). + status_code (Optional[Union[int, str]]): HTTP status code or gRPC status name. + kind (str): Transport kind identifier ("grpc", "rest", "rest_asyncio", or "http"). + http_method (Optional[str]): HTTP request method (e.g. "GET", "POST"). + exc (Optional[BaseException]): Exception raised or returned during the attempt. + client_options (Optional[Any]): Client options used for feature gating. + logger (Optional[logging.Logger]): Explicit logger override. + status_message (Optional[str]): Explicit status message fallback. + """ + if not client_logging.is_error_logging_enabled(client_options): + return + + norm_status = str(status_code) if status_code is not None else None + if not _is_failed_attempt(norm_status, exc): + return + + try: + rpc_system = _resolve_rpc_system(kind) + extra_payload, resolved_status = _build_attempt_log_extra( + span=span, + target=target, + status_code=status_code, + norm_status=norm_status, + kind=rpc_system, + http_method=http_method, + exc=exc, + status_message=status_message, + ) + target_logger = logger or _get_service_logger( + target if rpc_system == "grpc" else None + ) + target_logger.debug( + "Transport attempt failed for %s (status=%s)", + target or "unknown_target", + resolved_status or "ERROR", + extra=extra_payload, + ) + except Exception: + # Fail-open: never disrupt the caller's RPC flow on logging errors + pass + + def _extract_endpoint_attributes( client_options: ClientOptions | dict[str, Any] | None = None, default_host: str | None = None, @@ -584,11 +810,149 @@ async def intercept_stream_stream( return await continuation(client_call_details, request_iterator) +class _BaseGrpcAttemptLoggingInterceptor: + """Shared base logic for sync and async gRPC attempt logging interceptors. + + Args: + client_options (Optional[Any]): Client options used for error logging gating. + """ + + __slots__ = ("_client_options",) + + def __init__(self, client_options: Any = None) -> None: + self._client_options = client_options + + @contextlib.contextmanager + def _capture_attempt_exception(self, method_str: str): + """Logs any unhandled exception raised during gRPC continuation dispatch and re-raises. + + Args: + method_str (str): Decoded gRPC method name. + """ + try: + yield + except BaseException as exc: + log_transport_attempt( + target=method_str, + kind="grpc", + exc=exc, + client_options=self._client_options, + ) + raise + + +class _GrpcAttemptLoggingInterceptor( + _BaseGrpcAttemptLoggingInterceptor, _SyncUnaryUnaryClientInterceptor +): + """Sync gRPC client interceptor that logs failed wire attempts at DEBUG level.""" + + __slots__ = () + + def intercept_unary_unary( + self, continuation: Any, client_call_details: Any, request: Any + ) -> Any: + """Intercepts a sync unary-unary gRPC call and logs failed wire attempts. + + Args: + continuation (Callable): Next interceptor or channel continuation callable. + client_call_details (Any): gRPC call details containing the method name. + request (Any): Outbound protobuf request message. + + Returns: + Any: The gRPC Call/Future response returned by ``continuation``. + """ + method_str = _decode_method_name(getattr(client_call_details, "method", None)) + with self._capture_attempt_exception(method_str): + response = continuation(client_call_details, request) + + exc_fn = getattr(response, "exception", None) + if callable(exc_fn): + resp_exc = exc_fn() + if resp_exc is not None: + code_fn = getattr(response, "code", None) + code = code_fn() if callable(code_fn) else None + code_name = getattr(code, "name", None) or str(code or "UNKNOWN") + log_transport_attempt( + target=method_str, + status_code=code_name, + kind="grpc", + exc=resp_exc, + client_options=self._client_options, + ) + return response + + +class _AsyncGrpcAttemptLoggingInterceptor( + _BaseGrpcAttemptLoggingInterceptor, _AsyncUnaryUnaryClientInterceptor +): + """AsyncIO gRPC client interceptor that logs failed wire attempts at DEBUG level.""" + + __slots__ = () + + async def intercept_unary_unary( + self, continuation: Any, client_call_details: Any, request: Any + ) -> Any: + """Intercepts an async unary-unary gRPC call and logs failed wire attempts. + + Args: + continuation (Callable): Next async interceptor or channel continuation callable. + client_call_details (Any): gRPC call details containing the method name. + request (Any): Outbound protobuf request message. + + Returns: + Any: The async gRPC call object returned by ``continuation``. + """ + method_str = _decode_method_name(getattr(client_call_details, "method", None)) + with self._capture_attempt_exception(method_str): + call = await continuation(client_call_details, request) + code_fn = getattr(call, "code", None) + if callable(code_fn): + code = await code_fn() + code_name = getattr(code, "name", str(code)) + if code_name != "OK": + details_fn = getattr(call, "details", None) + details = await details_fn() if callable(details_fn) else "" + log_transport_attempt( + target=method_str, + status_code=code_name, + kind="grpc", + exc=Exception(details) if details else None, + client_options=self._client_options, + status_message=details or None, + ) + return call + + +def _build_otel_grpc_hook_kwargs( + client_options: ClientOptions | dict[str, Any] | None = None, + default_host: str | None = None, +) -> dict[str, Any]: + """Builds shared keyword arguments for upstream opentelemetry-instrumentation-grpc. + + Args: + client_options (Optional[Union[ClientOptions, dict[str, Any]]]): Client options + used to resolve endpoint attributes and custom tracer provider. + default_host (Optional[str]): Fallback transport host when ``api_endpoint`` is not + set on ``client_options``. + + Returns: + dict[str, Any]: Keyword arguments (`tracer_provider`, `request_hook`, `response_hook`). + """ + endpoint_attrs = _extract_endpoint_attributes( + client_options, default_host=default_host + ) + return { + "tracer_provider": _get_tracer_provider(client_options), + "request_hook": _make_grpc_client_request_hook(endpoint_attrs), + "response_hook": _grpc_client_response_hook, + } + + def get_otel_interceptor( client_options: ClientOptions | dict[str, Any] | None = None, default_host: str | None = None, ) -> Callable[[grpc.Channel], grpc.Channel] | None: - """Returns an interceptor callable that wraps a sync gRPC channel with OpenTelemetry tracing. + """Returns a callable that wraps a sync gRPC channel with tracing and/or error logging. Args: client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options @@ -597,44 +961,57 @@ def get_otel_interceptor( set on ``client_options``. Returns: - Callable[[grpc.Channel], grpc.Channel] | None: An interceptor callable if OpenTelemetry - tracing is enabled and installed, None otherwise. + Callable[[grpc.Channel], grpc.Channel] | None: A channel-wrapping callable if + OpenTelemetry tracing or error logging is enabled, None otherwise. """ - if not is_otel_capabilities_enabled(client_options): + tracing_enabled = is_otel_capabilities_enabled(client_options) + error_logging_enabled = client_logging.is_error_logging_enabled(client_options) + if not tracing_enabled and not error_logging_enabled: return None - import opentelemetry.instrumentation.grpc as otel_grpc # type: ignore[import-not-found] + custom_interceptors: list[Any] = [] + tracing_interceptor: Any = None - endpoint_attrs = _extract_endpoint_attributes( - client_options, default_host=default_host - ) - request_hook = _make_grpc_client_request_hook(endpoint_attrs) + if tracing_enabled: + import opentelemetry.instrumentation.grpc as otel_grpc # type: ignore[import-not-found] - interceptor: ClientInterceptor = otel_grpc.client_interceptor( - tracer_provider=_get_tracer_provider(client_options), - request_hook=request_hook, - response_hook=_grpc_client_response_hook, - ) - suppressor = _SuppressingClientInterceptor() + tracing_interceptor = otel_grpc.client_interceptor( + **_build_otel_grpc_hook_kwargs(client_options, default_host) + ) + custom_interceptors.append(_SuppressingClientInterceptor()) + + if error_logging_enabled: + custom_interceptors.append(_GrpcAttemptLoggingInterceptor(client_options)) + + def apply_sync_otel_interceptors(channel: grpc.Channel) -> grpc.Channel: + """Applies configured sync observability interceptors to a gRPC channel. + + Args: + channel (grpc.Channel): The synchronous gRPC channel to wrap. - def otel_interceptor(channel: grpc.Channel) -> grpc.Channel: - if grpc is not None: + Returns: + grpc.Channel: The intercepted gRPC channel. + """ + if grpc is not None and custom_interceptors: try: - channel = grpc.intercept_channel(channel, suppressor) + channel = grpc.intercept_channel(channel, *custom_interceptors) except (AttributeError, TypeError): # Fallback if grpc lacks intercept_channel or custom channel type rejects it. pass - return otel_grpc.intercept_channel(channel, interceptor) + if tracing_interceptor is not None: + return otel_grpc.intercept_channel(channel, tracing_interceptor) + return channel - otel_interceptor._is_otel_interceptor = True # type: ignore[attr-defined] - return otel_interceptor + # Ignored by mypy: Function attribute used to tag our interceptor wrapper + apply_sync_otel_interceptors._is_otel_interceptor = True # type: ignore[attr-defined] + return apply_sync_otel_interceptors def get_otel_async_interceptor( client_options: ClientOptions | dict[str, Any] | None = None, default_host: str | None = None, ) -> Sequence[grpc.aio.ClientInterceptor] | None: - """Returns async gRPC client interceptors for OpenTelemetry tracing. + """Returns async gRPC client interceptors for OpenTelemetry tracing and/or error logging. Args: client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options @@ -643,31 +1020,31 @@ def get_otel_async_interceptor( set on ``client_options``. Returns: - Sequence[grpc.aio.ClientInterceptor] | None: Instantiated OpenTelemetry async - client interceptors if tracing is enabled and installed, None otherwise. + Sequence[grpc.aio.ClientInterceptor] | None: Instantiated async client interceptors + if OpenTelemetry tracing or error logging is enabled, None otherwise. """ - if not is_otel_capabilities_enabled(client_options): + tracing_enabled = is_otel_capabilities_enabled(client_options) + error_logging_enabled = client_logging.is_error_logging_enabled(client_options) + if not tracing_enabled and not error_logging_enabled: return None - # Ignored by mypy: Optional dependency only loaded if early-return is skipped - import opentelemetry.instrumentation.grpc as otel_grpc # type: ignore[import-not-found] + interceptors: list[Any] = [] + if tracing_enabled: + # Ignored by mypy: Optional dependency only loaded if tracing_enabled is True + import opentelemetry.instrumentation.grpc as otel_grpc # type: ignore[import-not-found] - endpoint_attrs = _extract_endpoint_attributes( - client_options, default_host=default_host - ) - request_hook = _make_grpc_client_request_hook(endpoint_attrs) + raw_interceptors = otel_grpc.aio_client_interceptors( + **_build_otel_grpc_hook_kwargs(client_options, default_host) + ) + interceptors.extend( + raw_interceptors + if isinstance(raw_interceptors, (list, tuple)) + else [raw_interceptors] + ) + interceptors.append(_AsyncSuppressingClientInterceptor()) - raw_interceptors = otel_grpc.aio_client_interceptors( - tracer_provider=_get_tracer_provider(client_options), - request_hook=request_hook, - response_hook=_grpc_client_response_hook, - ) - interceptors = ( - list(raw_interceptors) - if isinstance(raw_interceptors, (list, tuple)) - else [raw_interceptors] - ) - interceptors.append(_AsyncSuppressingClientInterceptor()) + if error_logging_enabled: + interceptors.append(_AsyncGrpcAttemptLoggingInterceptor(client_options)) return interceptors @@ -900,12 +1277,65 @@ def __init__( self._parent_span: Any = None self._cm: Any = None + def _log_http_attempt( + self, + status_code: int | str | None = None, + exc: BaseException | None = None, + status_message: str | None = None, + ) -> None: + """Emits a DEBUG transport attempt log for a failed HTTP request (fail-open). + + Args: + status_code (Optional[Union[int, str]]): HTTP response status code or error code. + exc (Optional[BaseException]): Exception raised during HTTP dispatch, if any. + status_message (Optional[str]): Extracted HTTP error message fallback. + """ + try: + resolved_url = ( + getattr(self._request, "url", None) + if self._request is not None + else self._url + ) + resolved_method = ( + getattr(self._request, "method", None) + if self._request is not None + else self._method + ) + log_transport_attempt( + span=self._span, + target=resolved_url, + status_code=status_code, + kind="http", + http_method=resolved_method, + exc=exc, + client_options=self._client_options, + status_message=status_message, + ) + except Exception: + pass + def record_response(self, response: Any) -> None: - """Record HTTP response attributes on the active span. + """Record HTTP response attributes on the active span and log failed attempts. Args: response (Any): The HTTP response object (e.g. requests.Response). """ + status_code = getattr( + response, "status_code", getattr(response, "status", None) + ) + code_int: int | None = None + error_msg: str | None = None + if status_code is not None: + try: + code_int = int(status_code) + if code_int >= 400: + error_msg = self._extract_http_error_message(response) + self._log_http_attempt( + status_code=code_int, status_message=error_msg + ) + except Exception: + pass + span = self._span if span is None or not hasattr(span, "set_attribute"): return @@ -916,11 +1346,7 @@ def record_response(self, response: Any) -> None: StatusCode, ) - status_code = getattr( - response, "status_code", getattr(response, "status", None) - ) - if status_code is not None: - code_int = int(status_code) + if code_int is not None: span.set_attribute("http.response.status_code", code_int) if self._parent_span is not None: self._parent_span.set_attribute( @@ -929,7 +1355,6 @@ def record_response(self, response: Any) -> None: if code_int >= 400: span.set_status(Status(StatusCode.ERROR)) span.set_attribute("error.type", str(code_int)) - error_msg = self._extract_http_error_message(response) if error_msg: span.set_attribute("status.message", error_msg) @@ -955,7 +1380,7 @@ def record_response(self, response: Any) -> None: record_http_response = record_response def record_error(self, exc: BaseException | None) -> None: - """Record supplemental Cloud Observability error attributes on the active span. + """Record supplemental Cloud Observability error attributes on the active span and log failed attempts. Upstream OpenTelemetry's ``start_as_current_span`` context manager automatically records the exception event and marks the span status as ERROR when an unhandled @@ -971,15 +1396,18 @@ def record_error(self, exc: BaseException | None) -> None: Args: exc (Optional[BaseException]): The exception raised during dispatch. """ + if exc is None: + return + + status_code = getattr(exc, "code", None) or getattr(exc, "status_code", None) + self._log_http_attempt(status_code=status_code, exc=exc) + span = self._span - if span is None or exc is None: + if span is None: return try: if hasattr(span, "set_attribute"): - status_code = getattr(exc, "code", None) or getattr( - exc, "status_code", None - ) if status_code: span.set_attribute("error.type", str(status_code)) else: @@ -1060,10 +1488,12 @@ def __enter__(self) -> "_TraceContext": def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: if self._cm is not None: try: - if exc_val is not None and self._span is not None: + if exc_val is not None: self.record_error(exc_val) finally: self._cm.__exit__(exc_type, exc_val, exc_tb) + elif exc_val is not None: + self.record_error(exc_val) # Always return None so caller exceptions are never suppressed return None diff --git a/packages/google-api-core/google/api_core/client_logging.py b/packages/google-api-core/google/api_core/client_logging.py index 53d9f3a5c574..68d1f8d18398 100644 --- a/packages/google-api-core/google/api_core/client_logging.py +++ b/packages/google-api-core/google/api_core/client_logging.py @@ -2,10 +2,35 @@ import logging import os from collections.abc import Mapping -from typing import List, Optional +from typing import Any, List, Optional + +from google.api_core import _feature_gating_helpers _LOGGING_INITIALIZED = False _BASE_LOGGER_NAME = "google" +_LOGGER_PROVIDER = "logger_provider" + + +def is_error_logging_enabled(client_options: Any = None) -> bool: + """Determines if OpenTelemetry-correlated Production Error Logging (PEL) is enabled. + + This feature gate controls action-oriented error and retry logging + (configured via ``GOOGLE_SDK_EXPERIMENTAL_PYTHON_LOGGING_ENABLED`` and + ``client_options.logger_provider``), and is distinct from Application Payload + Debug Logging (APDL) configured via ``GOOGLE_SDK_PYTHON_LOGGING_SCOPE``. + + Args: + client_options (Optional[Any]): The client options object or dictionary. + + Returns: + bool: True if production error logging is enabled, False otherwise. + """ + return _feature_gating_helpers.resolve_feature_flags( + env_var="GOOGLE_SDK_EXPERIMENTAL_PYTHON_LOGGING_ENABLED", + feature_key=_LOGGER_PROVIDER, + configuration=client_options, + ) + # Fields to be included in the StructuredLogFormatter. # diff --git a/packages/google-api-core/google/api_core/gapic_v1/method.py b/packages/google-api-core/google/api_core/gapic_v1/method.py index 2dd3073da70d..5f30369d1e19 100644 --- a/packages/google-api-core/google/api_core/gapic_v1/method.py +++ b/packages/google-api-core/google/api_core/gapic_v1/method.py @@ -18,12 +18,18 @@ compression, pagination, and long-running operations to gRPC methods. """ +import asyncio import contextlib import enum import functools -from typing import List, Tuple - -from google.api_core import _observability, grpc_helpers +import logging +from typing import Any, List, Optional, Tuple + +from google.api_core import ( + _observability, + client_logging, + grpc_helpers, +) from google.api_core.gapic_v1 import client_info from google.api_core.gapic_v1.client_info import METRICS_METADATA_KEY from google.api_core.timeout import TimeToDeadlineTimeout @@ -110,22 +116,6 @@ def _extract_metrics_header(metadata) -> Tuple[str, List[Tuple[str, str]]]: return metric_str, arbitrary_metadata -def _extract_rpc_identity( - method_name: str, -) -> Tuple[str, str, str]: - """Extract (full_rpc_name, service_name, rpc_method_name) from an explicit method name. - - Args: - method_name: Explicit RPC name (e.g. "/google.cloud.secretmanager.v1.SecretManagerService/AccessSecretVersion"). - - Returns: - Tuple[str, str, str]: A 3-tuple of (full_rpc_name, service_name, rpc_method_name). - """ - method_str = method_name.lstrip("/") - service, _, method = method_str.rpartition("/") - return method_str, service, method - - class _GapicCallable(object): """Callable that applies retry, timeout, and metadata logic. @@ -181,6 +171,9 @@ def __init__( self._retry = retry self._timeout = timeout self._compression = compression + self._client_options = client_options + self._method_name = method_name + self._kind = kind # Pre-extract the x-goog-api-client header from the initialized metadata. self._x_goog_api_client, remaining = _extract_metrics_header(metadata) @@ -215,12 +208,12 @@ def __init__( else: tracer = trace.get_tracer("google.api_core") - span_name, _, _ = _extract_rpc_identity(method_name) + span_name, _, _ = _observability.extract_rpc_identity(method_name) endpoint_attrs = _observability._extract_endpoint_attributes( client_options, default_host=default_host ) span_attributes = { - "rpc.system.name": "http" if kind.startswith("rest") else "grpc", + "rpc.system.name": _observability._resolve_rpc_system(kind), "rpc.method": span_name, **endpoint_attrs, } @@ -270,8 +263,101 @@ def _prepare_call(self, timeout, retry, compression, kwargs): return wrapped_func + def _emit_method_error_log( + self, + exc: BaseException, + span: Any, + error_attrs: dict[str, Any], + is_cancelled: bool, + logger: Optional[logging.Logger] = None, + ) -> None: + """Constructs and emits the L3 terminal method error/cancellation log record. + + Args: + exc (BaseException): The terminal exception raised by the RPC method. + span (Any): The active Tier 3 OpenTelemetry span, if any. + error_attrs (dict[str, Any]): Pre-extracted telemetry error attributes. + is_cancelled (bool): True if the RPC was cancelled rather than failed. + logger (Optional[logging.Logger]): Explicit logger override for testing. + """ + try: + rpc_system = _observability._resolve_rpc_system(self._kind) + if is_cancelled: + error_attrs.pop("exception.stacktrace", None) + extra_payload: dict[str, Any] = { + "rpcName": self._method_name, + "rpc.system.name": rpc_system, + **error_attrs, + **_observability.extract_trace_context(span), + } + extra_payload["json_fields"] = dict(extra_payload) + + log_target = logger or _observability._get_service_logger( + self._method_name, self._target + ) + log_level = logging.DEBUG if is_cancelled else logging.ERROR + log_target.log( + log_level, + ( + "RPC cancelled for %s: %s" + if is_cancelled + else "RPC failed for %s: %s" + ), + self._method_name or "unknown_rpc", + exc, + exc_info=not is_cancelled, + extra=extra_payload, + ) + except Exception: + pass + + def _report_method_error( + self, + exc: BaseException, + span: Any = None, + logger: Optional[logging.Logger] = None, + ) -> None: + """Enriches the active Tier 3 method span and emits an L3 terminal method error log. + + Args: + exc (BaseException): The terminal exception raised by the RPC method. + span (Optional[Any]): The active Tier 3 OpenTelemetry span, if any. + logger (Optional[logging.Logger]): Explicit logger override for testing. + """ + error_logging_enabled = client_logging.is_error_logging_enabled( + self._client_options + ) + if span is None and not error_logging_enabled: + return + + error_attrs = _observability.build_telemetry_error_attributes( + exc, include_exception=error_logging_enabled + ) + status_code = error_attrs.get("rpc.response.status_code", "") + is_cancelled = status_code in ("CANCELLED", "CancelledError") or isinstance( + exc, asyncio.CancelledError + ) + if is_cancelled: + error_attrs["rpc.response.status_code"] = "CANCELLED" + + # 1. Enrich active Tier 3 method span + if span is not None and hasattr(span, "set_attribute"): + try: + for k, v in error_attrs.items(): + if not k.startswith("exception."): + span.set_attribute(k, v) + except Exception: + pass + + # 2. Emit L2/L3 Terminal Method Error Log + if error_logging_enabled: + self._emit_method_error_log( + exc, span, error_attrs, is_cancelled, logger=logger + ) + @contextlib.contextmanager def _trace_span(self): + """Context manager that wraps method execution in a Tier 3 OpenTelemetry span.""" span_cm = contextlib.nullcontext() if self._start_span_fn is not None: try: @@ -288,11 +374,7 @@ def _trace_span(self): except (KeyboardInterrupt, SystemExit, GeneratorExit): raise except BaseException as exc: - if span is not None and hasattr(span, "set_attribute"): - for k, v in _observability.build_telemetry_error_attributes( - exc - ).items(): - span.set_attribute(k, v) + self._report_method_error(exc, span) raise def __call__( diff --git a/packages/google-api-core/tests/asyncio/gapic/test_method_async.py b/packages/google-api-core/tests/asyncio/gapic/test_method_async.py index 0108d9e8417a..10aeeb6f13c4 100644 --- a/packages/google-api-core/tests/asyncio/gapic/test_method_async.py +++ b/packages/google-api-core/tests/asyncio/gapic/test_method_async.py @@ -630,3 +630,63 @@ async def test_invoke_wrapped_method_with_metadata_and_no_client_info(): method.assert_called_once_with( mock.sentinel.request, metadata=[("custom-header", "val")] ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "exc_factory,expected_level_name,expected_exc_info,expected_status,expect_stacktrace", + [ + pytest.param( + lambda: exceptions.NotFound("Secret not found"), + "ERROR", + True, + "NOT_FOUND", + True, + id="terminal_error", + ), + pytest.param( + lambda: asyncio.CancelledError("Task cancelled"), + "DEBUG", + False, + "CANCELLED", + False, + id="cancelled_error", + ), + ], +) +async def test_wrap_method_async_logging_error_and_cancelled( + monkeypatch, + mock_otel, + exc_factory, + expected_level_name, + expected_exc_info, + expected_status, + expect_stacktrace, +): + """Proves that _AsyncGapicCallable emits ERROR on terminal failure and DEBUG without traceback on asyncio.CancelledError.""" + import logging + + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_LOGGING_ENABLED", "true") + mock_logger = mock.Mock() + err = exc_factory() + mock_target = mock.AsyncMock(side_effect=err) + wrapped = gapic_v1.method_async.wrap_method( + mock_target, + method_name="/google.cloud.secretmanager.v1.SecretManagerService/GetSecret", + ) + with mock.patch( + "google.api_core._observability._get_service_logger", + return_value=mock_logger, + ): + with pytest.raises(type(err)): + await wrapped() + + mock_otel.span.set_attribute.assert_any_call( + "rpc.response.status_code", expected_status + ) + mock_logger.log.assert_called_once() + args, kwargs = mock_logger.log.call_args + assert args[0] == getattr(logging, expected_level_name) + assert kwargs["exc_info"] is expected_exc_info + assert kwargs["extra"]["rpc.response.status_code"] == expected_status + assert ("exception.stacktrace" in kwargs["extra"]) is expect_stacktrace diff --git a/packages/google-api-core/tests/unit/gapic/test_method.py b/packages/google-api-core/tests/unit/gapic/test_method.py index 292f05a73b50..3e0402dcbd41 100644 --- a/packages/google-api-core/tests/unit/gapic/test_method.py +++ b/packages/google-api-core/tests/unit/gapic/test_method.py @@ -849,3 +849,84 @@ def test_wrap_method_uninstrumented_exception(): with pytest.raises(RuntimeError, match="uninstrumented error"): wrapped() + + +def test_gapic_callable_report_method_error_and_logging(monkeypatch): + """Proves _GapicCallable._report_method_error enriches spans, emits ERROR logs on failure, and is silent on success.""" + import logging + import types + + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_LOGGING_ENABLED", "true") + + mock_ok_target = mock.Mock(return_value="ok") + wrapped_ok = google.api_core.gapic_v1.method.wrap_method( + mock_ok_target, + method_name="/google.cloud.secretmanager.v1.SecretManagerService/GetSecret", + ) + mock_logger = mock.Mock() + with mock.patch( + "google.api_core._observability._get_service_logger", + return_value=mock_logger, + ): + assert wrapped_ok() == "ok" + mock_logger.log.assert_not_called() + + error_info = types.SimpleNamespace( + domain="googleapis.com", + reason="SECRET_NOT_FOUND", + metadata={"project": "my-proj"}, + ) + err = exceptions.NotFound("Secret missing") + err._error_info = error_info + + wrapped_fail = google.api_core.gapic_v1.method.wrap_method( + mock.Mock(side_effect=err), + method_name="/google.cloud.secretmanager.v1.SecretManagerService/GetSecret", + kind="rest", + ) + mock_span = mock.Mock() + mock_span.get_span_context.return_value = types.SimpleNamespace( + is_valid=True, + trace_id=0x0123456789ABCDEF0123456789ABCDEF, + span_id=0x0123456789ABCDEF, + trace_flags=0x01, + ) + + err_logger = mock.Mock() + wrapped_fail._report_method_error(err, span=mock_span, logger=err_logger) + err_logger.log.assert_called_once() + args, kwargs = err_logger.log.call_args + assert args[0] == logging.ERROR + assert kwargs["exc_info"] is True + extra = kwargs["extra"] + assert ( + extra["rpcName"] + == "/google.cloud.secretmanager.v1.SecretManagerService/GetSecret" + ) + assert extra["rpc.system.name"] == "http" + assert extra["rpc.response.status_code"] == "NOT_FOUND" + assert extra["error.type"] == "SECRET_NOT_FOUND" + assert extra["gcp.errors.domain"] == "googleapis.com" + assert extra["gcp.errors.metadata.project"] == "my-proj" + assert extra["status.message"] == "Secret missing" + assert extra["exception.type"] == "NotFound" + assert "exception.stacktrace" in extra + assert extra["trace_id"] == "0123456789abcdef0123456789abcdef" + assert extra["span_id"] == "0123456789abcdef" + assert extra["json_fields"]["error.type"] == "SECRET_NOT_FOUND" + + cancel_logger = mock.Mock() + wrapped_fail._report_method_error( + exceptions.Cancelled("Call cancelled"), span=mock_span, logger=cancel_logger + ) + cancel_logger.log.assert_called_once() + c_args, c_kwargs = cancel_logger.log.call_args + assert c_args[0] == logging.DEBUG + assert c_kwargs["exc_info"] is False + assert "exception.stacktrace" not in c_kwargs["extra"] + assert c_kwargs["extra"]["rpc.response.status_code"] == "CANCELLED" + + bad_span = mock.Mock() + bad_span.set_attribute.side_effect = RuntimeError("span broken") + err_logger.log.side_effect = RuntimeError("logger broken") + wrapped_fail._report_method_error(err, span=bad_span, logger=err_logger) diff --git a/packages/google-api-core/tests/unit/test_client_logging.py b/packages/google-api-core/tests/unit/test_client_logging.py index bcecc98d1ecc..0ad2004c469d 100644 --- a/packages/google-api-core/tests/unit/test_client_logging.py +++ b/packages/google-api-core/tests/unit/test_client_logging.py @@ -223,3 +223,21 @@ def test_structured_log_formatter_json_fields_only(): "gcp.errors.metadata.service", "custom_from_json_fields", ] + + +def test_is_error_logging_enabled(monkeypatch): + import pytest + + from google.api_core._feature_gating_helpers import FeatureGatingError + from google.api_core.client_logging import is_error_logging_enabled + + monkeypatch.delenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_LOGGING_ENABLED", raising=False) + assert is_error_logging_enabled() is False + + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_LOGGING_ENABLED", "true") + assert is_error_logging_enabled() is True + + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_LOGGING_ENABLED", "false") + assert is_error_logging_enabled() is False + with pytest.raises(FeatureGatingError): + is_error_logging_enabled({"logger_provider": object()}) diff --git a/packages/google-api-core/tests/unit/test_observability.py b/packages/google-api-core/tests/unit/test_observability.py index ff0e71627aba..39a833c165c7 100644 --- a/packages/google-api-core/tests/unit/test_observability.py +++ b/packages/google-api-core/tests/unit/test_observability.py @@ -2338,3 +2338,408 @@ def test_build_telemetry_error_attributes_include_exception( assert expected_exc_msg_substr in attrs_inc["exception.message"] assert "Traceback (most recent call last):" in attrs_inc["exception.stacktrace"] assert "raise exc_factory()" in attrs_inc["exception.stacktrace"] + + +@pytest.mark.parametrize( + "method_name,target,expected_name", + [ + pytest.param( + "/google.cloud.secretmanager.v1.SecretManagerService/GetSecret", + types.SimpleNamespace( + __module__="google.cloud.secretmanager_v1.services.secret_manager_service.transports.rest" + ), + "google.cloud.secretmanager_v1.services.secret_manager_service.transports.rest", + id="target_module_takes_precedence", + ), + pytest.param( + "/google.cloud.secretmanager.v1.SecretManagerService/GetSecret", + None, + "google.cloud.secretmanager.v1", + id="str_method_name_fallback", + ), + pytest.param( + b"/google.showcase.v1beta1.Echo/Echo", + None, + "google.showcase.v1beta1", + id="bytes_method_name_fallback", + ), + pytest.param( + "https://example.com/api", + None, + _observability._LOGGER.name, + id="non_google_url_fallback_to_module_logger", + ), + pytest.param( + None, + None, + _observability._LOGGER.name, + id="none_inputs_fallback_to_module_logger", + ), + ], +) +def test_get_service_logger(method_name, target, expected_name): + """Proves that _get_service_logger resolves target module, RPC method name, or fallback logger.""" + resolved = _observability._get_service_logger(method_name, target) + assert resolved.name == expected_name + + +@pytest.mark.parametrize( + "logging_enabled,kwargs", + [ + pytest.param( + "false", + { + "target": "/google.showcase.v1beta1.Echo/Echo", + "status_code": "UNAVAILABLE", + "kind": "grpc", + "exc": RuntimeError("down"), + }, + id="logging_disabled", + ), + pytest.param( + "true", + { + "target": "/google.showcase.v1beta1.Echo/Echo", + "status_code": "OK", + "kind": "grpc", + }, + id="grpc_ok_status", + ), + pytest.param( + "true", + { + "target": "https://example.com/v1/echo", + "status_code": 200, + "kind": "http", + "http_method": "POST", + }, + id="http_200_status", + ), + ], +) +def test_log_transport_attempt_no_op(monkeypatch, logging_enabled, kwargs): + """Proves that log_transport_attempt is a no-op when disabled or when an attempt succeeds.""" + monkeypatch.setenv( + "GOOGLE_SDK_EXPERIMENTAL_PYTHON_LOGGING_ENABLED", logging_enabled + ) + mock_logger = mock.Mock() + _observability.log_transport_attempt(logger=mock_logger, **kwargs) + mock_logger.debug.assert_not_called() + + +@pytest.mark.parametrize( + "kwargs,expected_extra", + [ + pytest.param( + { + "target": "/google.showcase.v1beta1.Echo/Echo", + "status_code": "UNAVAILABLE", + "kind": "grpc", + "exc": RuntimeError("transient error"), + }, + { + "rpc.system.name": "grpc", + "rpc.method": "/google.showcase.v1beta1.Echo/Echo", + "rpc.response.status_code": "UNAVAILABLE", + "error.type": "UNAVAILABLE", + "status.message": "transient error", + "exception.type": "RuntimeError", + "trace_id": "0123456789abcdef0123456789abcdef", + "span_id": "0123456789abcdef", + }, + id="failed_grpc_attempt_with_span", + ), + pytest.param( + { + "target": "https://example.com/v1/echo", + "status_code": 503, + "kind": "http", + "http_method": "POST", + }, + { + "rpc.system.name": "http", + "url.full": "https://example.com/v1/echo", + "httpRequest": { + "requestMethod": "POST", + "requestUrl": "https://example.com/v1/echo", + "status": 503, + }, + "error.type": "503", + "status.message": "HTTP 503", + }, + id="failed_http_attempt_without_exception", + ), + ], +) +def test_log_transport_attempt_emits_debug(monkeypatch, kwargs, expected_extra): + """Proves that log_transport_attempt emits structured DEBUG logs on failed wire attempts.""" + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_LOGGING_ENABLED", "true") + valid_ctx = types.SimpleNamespace( + is_valid=True, + trace_id=0x0123456789ABCDEF0123456789ABCDEF, + span_id=0x0123456789ABCDEF, + trace_flags=0x01, + ) + span = mock.Mock() + span.get_span_context.return_value = valid_ctx + mock_logger = mock.Mock() + + _observability.log_transport_attempt(span=span, logger=mock_logger, **kwargs) + + mock_logger.debug.assert_called_once() + extra = mock_logger.debug.call_args[1]["extra"] + for key, val in expected_extra.items(): + assert extra[key] == val + assert extra["json_fields"][key] == val + + # Verify fail-open resilience if logger.debug raises + mock_logger.debug.side_effect = RuntimeError("logger failed") + _observability.log_transport_attempt(span=span, logger=mock_logger, **kwargs) + + +def test_grpc_attempt_logging_interceptor_sync_responses(monkeypatch): + """Proves that _GrpcAttemptLoggingInterceptor logs only failed sync gRPC response objects.""" + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_LOGGING_ENABLED", "true") + interceptor = _observability._GrpcAttemptLoggingInterceptor() + call_details = types.SimpleNamespace(method=b"/google.showcase.v1beta1.Echo/Echo") + + ok_response = mock.Mock() + ok_response.exception.return_value = None + assert ( + interceptor.intercept_unary_unary(lambda d, r: ok_response, call_details, "req") + is ok_response + ) + plain_response = object() + assert ( + interceptor.intercept_unary_unary( + lambda d, r: plain_response, call_details, "req" + ) + is plain_response + ) + + failed_response = mock.Mock() + failed_response.exception.return_value = RuntimeError("wire failure") + failed_response.code.return_value = types.SimpleNamespace(name="UNAVAILABLE") + + with mock.patch.object(_observability, "log_transport_attempt") as mock_log: + res = interceptor.intercept_unary_unary( + lambda d, r: failed_response, call_details, "req" + ) + assert res is failed_response + mock_log.assert_called_once() + assert mock_log.call_args[1]["target"] == "/google.showcase.v1beta1.Echo/Echo" + assert mock_log.call_args[1]["status_code"] == "UNAVAILABLE" + + +class _FakeSyncRpcError(Exception): + def code(self): + return types.SimpleNamespace(name="INTERNAL") + + +class _BadCodeSyncRpcError(Exception): + def code(self): + raise RuntimeError("bad code") + + +@pytest.mark.parametrize( + "exc_cls,expected_status", + [ + (_FakeSyncRpcError, "INTERNAL"), + (_BadCodeSyncRpcError, "_BadCodeSyncRpcError"), + ], +) +def test_grpc_attempt_logging_interceptor_sync_exceptions( + monkeypatch, exc_cls, expected_status +): + """Proves that _GrpcAttemptLoggingInterceptor logs exceptions raised directly by sync continuation.""" + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_LOGGING_ENABLED", "true") + interceptor = _observability._GrpcAttemptLoggingInterceptor() + call_details = types.SimpleNamespace(method=b"/google.showcase.v1beta1.Echo/Echo") + mock_logger = mock.Mock() + + def raising_continuation(d, r): + raise exc_cls("sync wire error") + + with mock.patch.object( + _observability, "_get_service_logger", return_value=mock_logger + ): + with pytest.raises(exc_cls): + interceptor.intercept_unary_unary(raising_continuation, call_details, "req") + mock_logger.debug.assert_called_once() + assert ( + mock_logger.debug.call_args[1]["extra"]["rpc.response.status_code"] + == expected_status + ) + + +@pytest.mark.asyncio +async def test_grpc_attempt_logging_interceptor_async(monkeypatch): + """Proves that _AsyncGrpcAttemptLoggingInterceptor logs failed async gRPC wire attempts.""" + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_LOGGING_ENABLED", "true") + interceptor = _observability._AsyncGrpcAttemptLoggingInterceptor() + call_details = types.SimpleNamespace(method=b"/google.showcase.v1beta1.Echo/Echo") + + ok_call = mock.AsyncMock() + ok_call.code.return_value = types.SimpleNamespace(name="OK") + assert ( + await interceptor.intercept_unary_unary( + mock.AsyncMock(return_value=ok_call), call_details, "req" + ) + is ok_call + ) + plain_call = object() + assert ( + await interceptor.intercept_unary_unary( + mock.AsyncMock(return_value=plain_call), call_details, "req" + ) + is plain_call + ) + + fake_call = mock.AsyncMock() + fake_call.code.return_value = types.SimpleNamespace(name="UNAVAILABLE") + fake_call.details.return_value = "backend unavailable" + + with mock.patch.object(_observability, "log_transport_attempt") as mock_log: + res = await interceptor.intercept_unary_unary( + mock.AsyncMock(return_value=fake_call), call_details, "req" + ) + assert res is fake_call + mock_log.assert_called_once() + assert mock_log.call_args[1]["target"] == "/google.showcase.v1beta1.Echo/Echo" + assert mock_log.call_args[1]["status_code"] == "UNAVAILABLE" + assert mock_log.call_args[1]["status_message"] == "backend unavailable" + + class FakeAsyncError(Exception): + def code(self): + raise ValueError("broken code") + + async def raising_continuation(d, r): + raise FakeAsyncError("async wire error") + + mock_logger = mock.Mock() + with mock.patch.object( + _observability, "_get_service_logger", return_value=mock_logger + ): + with pytest.raises(FakeAsyncError): + await interceptor.intercept_unary_unary( + raising_continuation, call_details, "req" + ) + mock_logger.debug.assert_called_once() + assert ( + mock_logger.debug.call_args[1]["extra"]["rpc.response.status_code"] + == "FakeAsyncError" + ) + + +@pytest.mark.parametrize( + "tracing_enabled,expected_async_len,expected_sync_arg_len", + [ + ("false", 1, 2), + ("true", 3, 3), + ], + ids=["logging_only", "tracing_and_logging"], +) +def test_get_otel_interceptors_with_logging_enabled( + monkeypatch, + mock_otel_grpc, + tracing_enabled, + expected_async_len, + expected_sync_arg_len, +): + """Proves that get_otel_interceptor and get_otel_async_interceptor attach logging interceptors.""" + monkeypatch.setenv( + "GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", tracing_enabled + ) + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_LOGGING_ENABLED", "true") + mock_otel_grpc.aio_client_interceptors.return_value = [mock.Mock()] + + sync_fn = _observability.get_otel_interceptor() + assert callable(sync_fn) + mock_channel = mock.Mock() + if _observability.grpc is not None: + with mock.patch.object( + _observability.grpc, "intercept_channel", return_value=mock_channel + ) as mock_ic: + sync_fn(mock_channel) + mock_ic.assert_called_once() + assert len(mock_ic.call_args[0]) == expected_sync_arg_len + with mock.patch.object( + _observability.grpc, "intercept_channel", side_effect=TypeError("bad") + ): + assert sync_fn(mock_channel) is not None + with mock.patch.object(_observability, "grpc", None): + assert sync_fn(mock_channel) is not None + + async_list = _observability.get_otel_async_interceptor() + assert len(async_list) == expected_async_len + assert any( + isinstance(i, _observability._AsyncGrpcAttemptLoggingInterceptor) + for i in async_list + ) + + +def test_trace_context_http_logging_on_error_status_and_exception(monkeypatch): + """Proves that _TraceContext logs HTTP error responses (>= 400) and dispatch exceptions when logging is enabled.""" + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "false") + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_LOGGING_ENABLED", "true") + + with mock.patch.object(_observability, "log_transport_attempt") as mock_log: + with _observability.trace_http_request( + method="POST", url="https://example.com/v1/echo" + ) as ctx: + ctx.record_response( + mock.Mock(status_code=503, headers={}, reason="Service Unavailable") + ) + mock_log.assert_called_once() + assert mock_log.call_args[1]["status_code"] == 503 + assert mock_log.call_args[1]["kind"] == "http" + assert mock_log.call_args[1]["status_message"] == "Service Unavailable" + + with mock.patch.object(_observability, "log_transport_attempt") as mock_log: + with pytest.raises(ConnectionError): + with _observability.trace_http_request( + method="GET", url="https://example.com/v1/echo" + ) as ctx: + ctx.record_error(None) + raise ConnectionError("reset") + mock_log.assert_called_once() + assert mock_log.call_args[1]["kind"] == "http" + assert isinstance(mock_log.call_args[1]["exc"], ConnectionError) + + with mock.patch.object( + _observability, "log_transport_attempt", side_effect=RuntimeError("log crash") + ): + ctx_err = _observability._TraceContext(method="GET", url="https://example.com") + ctx_err.record_error(ConnectionError("fail-open")) + ctx_err.record_response(mock.Mock(status_code=500, headers={})) + ctx_err.record_response(mock.Mock(status_code="invalid", headers={})) + + class BadTextResponse: + status_code = 500 + headers: dict = {} + + @property + def text(self): + raise RuntimeError("broken text") + + reason = "Fallback Reason" + + assert ( + _observability._TraceContext._extract_http_error_message(BadTextResponse()) + == "Fallback Reason" + ) + + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true") + mock_parent_span = mock.MagicMock() + mock_parent_span.is_recording.return_value = True + mock_parent_span.attributes = {} + mock_parent_span.set_attribute.side_effect = RuntimeError("parent attr crash") + mock_otel = mock.MagicMock() + mock_otel.trace.get_current_span.return_value = mock_parent_span + monkeypatch.setitem(sys.modules, "opentelemetry", mock_otel) + monkeypatch.setitem(sys.modules, "opentelemetry.trace", mock_otel.trace) + monkeypatch.setitem(sys.modules, "opentelemetry.instrumentation.grpc", mock.Mock()) + with _observability.trace_http_request( + method="GET", url="https://example.com/v1/echo", url_template="/v1/echo" + ): + pass From 44b088c062a12732714fd952777aa02289e06595 Mon Sep 17 00:00:00 2001 From: chalmer lowe Date: Thu, 5 Nov 2026 13:06:15 -0500 Subject: [PATCH 2/4] fix(api_core): normalize enum status codes and fail-open gRPC post-dispatch inspection --- .../google/api_core/_observability.py | 45 ++++++++++++------- .../tests/unit/test_observability.py | 27 +++++++++++ 2 files changed, 56 insertions(+), 16 deletions(-) diff --git a/packages/google-api-core/google/api_core/_observability.py b/packages/google-api-core/google/api_core/_observability.py index ef1d4faa840d..16d1c50b2724 100644 --- a/packages/google-api-core/google/api_core/_observability.py +++ b/packages/google-api-core/google/api_core/_observability.py @@ -484,7 +484,11 @@ def log_transport_attempt( if not client_logging.is_error_logging_enabled(client_options): return - norm_status = str(status_code) if status_code is not None else None + norm_status = ( + getattr(status_code, "name", str(status_code)) + if status_code is not None + else None + ) if not _is_failed_attempt(norm_status, exc): return @@ -865,20 +869,24 @@ def intercept_unary_unary( with self._capture_attempt_exception(method_str): response = continuation(client_call_details, request) - exc_fn = getattr(response, "exception", None) - if callable(exc_fn): - resp_exc = exc_fn() - if resp_exc is not None: - code_fn = getattr(response, "code", None) - code = code_fn() if callable(code_fn) else None - code_name = getattr(code, "name", None) or str(code or "UNKNOWN") - log_transport_attempt( - target=method_str, - status_code=code_name, - kind="grpc", - exc=resp_exc, - client_options=self._client_options, - ) + try: + exc_fn = getattr(response, "exception", None) + if callable(exc_fn): + resp_exc = exc_fn() + if resp_exc is not None: + code_fn = getattr(response, "code", None) + code = code_fn() if callable(code_fn) else None + code_name = getattr(code, "name", None) or str(code or "UNKNOWN") + log_transport_attempt( + target=method_str, + status_code=code_name, + kind="grpc", + exc=resp_exc, + client_options=self._client_options, + ) + except Exception: + # Fail-open: never disrupt the caller's RPC flow on post-dispatch inspection errors + pass return response @@ -905,6 +913,8 @@ async def intercept_unary_unary( method_str = _decode_method_name(getattr(client_call_details, "method", None)) with self._capture_attempt_exception(method_str): call = await continuation(client_call_details, request) + + try: code_fn = getattr(call, "code", None) if callable(code_fn): code = await code_fn() @@ -920,7 +930,10 @@ async def intercept_unary_unary( client_options=self._client_options, status_message=details or None, ) - return call + except Exception: + # Fail-open: never disrupt the caller's RPC flow on post-dispatch inspection errors + pass + return call def _build_otel_grpc_hook_kwargs( diff --git a/packages/google-api-core/tests/unit/test_observability.py b/packages/google-api-core/tests/unit/test_observability.py index 39a833c165c7..74d488f3d70e 100644 --- a/packages/google-api-core/tests/unit/test_observability.py +++ b/packages/google-api-core/tests/unit/test_observability.py @@ -2405,6 +2405,15 @@ def test_get_service_logger(method_name, target, expected_name): }, id="grpc_ok_status", ), + pytest.param( + "true", + { + "target": "/google.showcase.v1beta1.Echo/Echo", + "status_code": types.SimpleNamespace(name="OK"), + "kind": "grpc", + }, + id="grpc_enum_ok_status", + ), pytest.param( "true", { @@ -2530,6 +2539,15 @@ def test_grpc_attempt_logging_interceptor_sync_responses(monkeypatch): assert mock_log.call_args[1]["target"] == "/google.showcase.v1beta1.Echo/Echo" assert mock_log.call_args[1]["status_code"] == "UNAVAILABLE" + cancelled_response = mock.Mock() + cancelled_response.exception.side_effect = RuntimeError("FutureCancelledError") + assert ( + interceptor.intercept_unary_unary( + lambda d, r: cancelled_response, call_details, "req" + ) + is cancelled_response + ) + class _FakeSyncRpcError(Exception): def code(self): @@ -2609,6 +2627,15 @@ async def test_grpc_attempt_logging_interceptor_async(monkeypatch): assert mock_log.call_args[1]["status_code"] == "UNAVAILABLE" assert mock_log.call_args[1]["status_message"] == "backend unavailable" + broken_code_call = mock.AsyncMock() + broken_code_call.code.side_effect = RuntimeError("broken code inspection") + assert ( + await interceptor.intercept_unary_unary( + mock.AsyncMock(return_value=broken_code_call), call_details, "req" + ) + is broken_code_call + ) + class FakeAsyncError(Exception): def code(self): raise ValueError("broken code") From 55449b0d8034f08a80cc88c39b5997d5e844fce8 Mon Sep 17 00:00:00 2001 From: chalmer lowe Date: Thu, 5 Nov 2026 17:03:25 -0500 Subject: [PATCH 3/4] refactor(api-core): tighten type annotations and unify PEP 604 syntax --- .../google/api_core/_observability.py | 155 ++++++++++-------- .../google/api_core/client_logging.py | 15 +- .../google/api_core/gapic_v1/method.py | 27 +-- 3 files changed, 114 insertions(+), 83 deletions(-) diff --git a/packages/google-api-core/google/api_core/_observability.py b/packages/google-api-core/google/api_core/_observability.py index 16d1c50b2724..cb675df84c3c 100644 --- a/packages/google-api-core/google/api_core/_observability.py +++ b/packages/google-api-core/google/api_core/_observability.py @@ -19,10 +19,12 @@ from __future__ import annotations import contextlib +import enum import functools import logging import traceback import urllib.parse +from collections.abc import Iterator from typing import TYPE_CHECKING, Any, Callable, Sequence from google.api_core import _feature_gating_helpers, client_logging, exceptions @@ -66,7 +68,7 @@ def is_otel_capabilities_enabled( """Checks if OTel capabilities are enabled and installed. Args: - client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options + client_options (ClientOptions | dict[str, Any] | None): The client options object or dictionary. env_var (str): The environment variable to check for enablement. @@ -98,7 +100,7 @@ def extract_trace_context(span: Any = None) -> dict[str, Any]: span context exists or if OpenTelemetry is not installed. Args: - span (Optional[Any]): The active OpenTelemetry span, if any. + span (Any | None): The active OpenTelemetry span, if any. Returns: dict[str, Any]: Trace and span correlation attributes. @@ -216,11 +218,11 @@ def build_telemetry_error_attributes( gcp.errors.metadata.*, status.message, and optional exception.* fields. Args: - exc (Optional[BaseException]): An exception (e.g. GoogleAPICallError + exc (BaseException | None): An exception (e.g. GoogleAPICallError or grpc.RpcError) or ErrorInfo object. - status_code (Optional[str]): Optional pre-extracted or HTTP status code + status_code (str | None): Optional pre-extracted or HTTP status code fallback when exc is None or lacks a status code. - status_message (Optional[str]): Optional status message fallback when + status_message (str | None): Optional status message fallback when exc is None or lacks a message. include_exception (bool): When True, also populates exception.type, exception.message, and exception.stacktrace. @@ -296,7 +298,7 @@ def _decode_method_name(method: str | bytes | None) -> str: """Normalizes a string or bytes RPC method name to a UTF-8 string. Args: - method (Optional[Union[str, bytes]]): Raw RPC method name from call details. + method (str | bytes | None): Raw RPC method name from call details. Returns: str: Decoded string method name (or empty string if None). @@ -312,7 +314,7 @@ def extract_rpc_identity( """Extracts (full_rpc_name, service_name, rpc_method_name) from a gRPC method path. Args: - method_name (Optional[Union[str, bytes]]): Explicit gRPC method path + method_name (str | bytes | None): Explicit gRPC method path (e.g. "/google.cloud.secretmanager.v1.SecretManagerService/GetSecret"). Returns: @@ -327,7 +329,7 @@ def _resolve_rpc_system(kind: str | None = "grpc") -> str: """Normalizes a transport kind ('grpc', 'rest', 'rest_asyncio', 'http') to an OTel rpc.system.name. Args: - kind (Optional[str]): The transport kind identifier. + kind (str | None): The transport kind identifier. Returns: str: "http" for REST/HTTP transports, or "grpc" otherwise. @@ -339,7 +341,7 @@ def _resolve_rpc_system(kind: str | None = "grpc") -> str: def _get_service_logger( method_name: str | bytes | None = None, - target: Any = None, + target: Callable[..., Any] | None = None, ) -> logging.Logger: """Resolves a service-scoped logger matching GOOGLE_SDK_PYTHON_LOGGING_SCOPE. @@ -351,8 +353,8 @@ def _get_service_logger( 3. Falls back to `logging.getLogger("google.api_core._observability")`. Args: - method_name (Optional[Union[str, bytes]]): Full gRPC method path. - target (Optional[Any]): The low-level RPC target callable. + method_name (str | bytes | None): Full gRPC method path. + target (Callable[..., Any] | None): The low-level RPC target callable. Returns: logging.Logger: The resolved service-scoped logger. @@ -383,8 +385,8 @@ def _is_failed_attempt( """Determines whether a transport attempt ended with an error or exception. Args: - norm_status (Optional[str]): Normalized status code string (e.g. "OK", "200", "503"). - exc (Optional[BaseException]): Exception raised or returned during the attempt. + norm_status (str | None): Normalized status code string (e.g. "OK", "200", "503"). + exc (BaseException | None): Exception raised or returned during the attempt. Returns: bool: True if the attempt failed, False if it succeeded. @@ -397,9 +399,9 @@ def _is_failed_attempt( def _build_attempt_log_extra( - span: Any, + span: opentelemetry.trace.Span | None, target: str | None, - status_code: int | str | None, + status_code: int | str | enum.Enum | None, norm_status: str | None, kind: str, http_method: str | None, @@ -409,17 +411,17 @@ def _build_attempt_log_extra( """Builds the structured extra dictionary and resolved status for a failed wire attempt. Args: - span (Any): Active OpenTelemetry span, if any. - target (Optional[str]): Full RPC method name (gRPC) or request URL (HTTP). - status_code (Optional[Union[int, str]]): Raw HTTP status code or gRPC status name. - norm_status (Optional[str]): Stringified status code if provided. + span (opentelemetry.trace.Span | None): Active OpenTelemetry span, if any. + target (str | None): Full RPC method name (gRPC) or request URL (HTTP). + status_code (int | str | enum.Enum | None): Raw HTTP status code or gRPC status name/enum. + norm_status (str | None): Stringified status code if provided. kind (str): Transport kind ("grpc", "rest", "rest_asyncio", or "http"). - http_method (Optional[str]): HTTP request method (e.g. "GET", "POST"). - exc (Optional[BaseException]): Exception raised or returned during the attempt. - status_message (Optional[str]): Explicit status message fallback. + http_method (str | None): HTTP request method (e.g. "GET", "POST"). + exc (BaseException | None): Exception raised or returned during the attempt. + status_message (str | None): Explicit status message fallback. Returns: - tuple[dict[str, Any], Optional[str]]: A 2-tuple of (extra_payload, resolved_status). + tuple[dict[str, Any], str | None]: A 2-tuple of (extra_payload, resolved_status). """ rpc_system = _resolve_rpc_system(kind) error_attrs = build_telemetry_error_attributes( @@ -455,13 +457,13 @@ def _build_attempt_log_extra( def log_transport_attempt( - span: Any = None, + span: opentelemetry.trace.Span | None = None, target: str | None = None, - status_code: int | str | None = None, + status_code: int | str | enum.Enum | None = None, kind: str = "grpc", http_method: str | None = None, exc: BaseException | None = None, - client_options: Any = None, + client_options: ClientOptions | dict[str, Any] | None = None, logger: logging.Logger | None = None, status_message: str | None = None, ) -> None: @@ -471,15 +473,16 @@ def log_transport_attempt( emitting logs to avoid hot-path allocations. Args: - span (Optional[Any]): Active OpenTelemetry span, if any. - target (Optional[str]): Full RPC method name (gRPC) or request URL (HTTP). - status_code (Optional[Union[int, str]]): HTTP status code or gRPC status name. + span (opentelemetry.trace.Span | None): Active OpenTelemetry span, if any. + target (str | None): Full RPC method name (gRPC) or request URL (HTTP). + status_code (int | str | enum.Enum | None): HTTP status code or gRPC status name/enum. kind (str): Transport kind identifier ("grpc", "rest", "rest_asyncio", or "http"). - http_method (Optional[str]): HTTP request method (e.g. "GET", "POST"). - exc (Optional[BaseException]): Exception raised or returned during the attempt. - client_options (Optional[Any]): Client options used for feature gating. - logger (Optional[logging.Logger]): Explicit logger override. - status_message (Optional[str]): Explicit status message fallback. + http_method (str | None): HTTP request method (e.g. "GET", "POST"). + exc (BaseException | None): Exception raised or returned during the attempt. + client_options (ClientOptions | dict[str, Any] | None): Client options + used for feature gating. + logger (logging.Logger | None): Explicit logger override. + status_message (str | None): Explicit status message fallback. """ if not client_logging.is_error_logging_enabled(client_options): return @@ -525,9 +528,9 @@ def _extract_endpoint_attributes( """Extracts server.address, server.port (if non-default), and url.domain from client options if present. Args: - client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options + client_options (ClientOptions | dict[str, Any] | None): The client options object or dictionary. - default_host (Optional[str]): Fallback transport host when ``api_endpoint`` is not + default_host (str | None): Fallback transport host when ``api_endpoint`` is not set on ``client_options``. Returns: @@ -579,7 +582,7 @@ def _make_grpc_client_request_hook( """Creates an OpenTelemetry gRPC client request hook with optional endpoint attributes. Args: - endpoint_attrs (Optional[dict[str, Any]]): Optional static endpoint attributes to attach + endpoint_attrs (dict[str, Any] | None): Optional static endpoint attributes to attach to every span. Returns: @@ -642,7 +645,7 @@ def _grpc_client_response_hook(span: Any, response: Any) -> None: modern ``rpc.response.status_code`` in future releases, this hook can be retired. Args: - span (Optional[Any]): The OpenTelemetry span. + span (Any | None): The OpenTelemetry span. response (Any): The gRPC response object or details. """ if span is None or not getattr(span, "is_recording", lambda: False)(): @@ -666,7 +669,7 @@ def _get_tracer_provider( """Extracts the OpenTelemetry tracer provider from client options if present. Args: - client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options + client_options (ClientOptions | dict[str, Any] | None): The client options object or dictionary. Returns: @@ -818,20 +821,26 @@ class _BaseGrpcAttemptLoggingInterceptor: """Shared base logic for sync and async gRPC attempt logging interceptors. Args: - client_options (Optional[Any]): Client options used for error logging gating. + client_options (ClientOptions | dict[str, Any] | None): Client options + used for error logging gating. """ __slots__ = ("_client_options",) - def __init__(self, client_options: Any = None) -> None: + def __init__( + self, client_options: ClientOptions | dict[str, Any] | None = None + ) -> None: self._client_options = client_options @contextlib.contextmanager - def _capture_attempt_exception(self, method_str: str): + def _capture_attempt_exception(self, method_str: str) -> Iterator[None]: """Logs any unhandled exception raised during gRPC continuation dispatch and re-raises. Args: method_str (str): Decoded gRPC method name. + + Yields: + None """ try: yield @@ -853,13 +862,16 @@ class _GrpcAttemptLoggingInterceptor( __slots__ = () def intercept_unary_unary( - self, continuation: Any, client_call_details: Any, request: Any + self, + continuation: Callable[..., Any], + client_call_details: grpc.ClientCallDetails, + request: Any, ) -> Any: """Intercepts a sync unary-unary gRPC call and logs failed wire attempts. Args: - continuation (Callable): Next interceptor or channel continuation callable. - client_call_details (Any): gRPC call details containing the method name. + continuation (Callable[..., Any]): Next interceptor or channel continuation callable. + client_call_details (grpc.ClientCallDetails): gRPC call details containing the method name. request (Any): Outbound protobuf request message. Returns: @@ -898,13 +910,16 @@ class _AsyncGrpcAttemptLoggingInterceptor( __slots__ = () async def intercept_unary_unary( - self, continuation: Any, client_call_details: Any, request: Any + self, + continuation: Callable[..., Any], + client_call_details: grpc.aio.ClientCallDetails, + request: Any, ) -> Any: """Intercepts an async unary-unary gRPC call and logs failed wire attempts. Args: - continuation (Callable): Next async interceptor or channel continuation callable. - client_call_details (Any): gRPC call details containing the method name. + continuation (Callable[..., Any]): Next async interceptor or channel continuation callable. + client_call_details (grpc.aio.ClientCallDetails): gRPC call details containing the method name. request (Any): Outbound protobuf request message. Returns: @@ -943,9 +958,9 @@ def _build_otel_grpc_hook_kwargs( """Builds shared keyword arguments for upstream opentelemetry-instrumentation-grpc. Args: - client_options (Optional[Union[ClientOptions, dict[str, Any]]]): Client options + client_options (ClientOptions | dict[str, Any] | None): Client options used to resolve endpoint attributes and custom tracer provider. - default_host (Optional[str]): Fallback transport host when ``api_endpoint`` is not + default_host (str | None): Fallback transport host when ``api_endpoint`` is not set on ``client_options``. Returns: @@ -968,9 +983,9 @@ def get_otel_interceptor( """Returns a callable that wraps a sync gRPC channel with tracing and/or error logging. Args: - client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options + client_options (ClientOptions | dict[str, Any] | None): The client options object or dictionary used for feature gating and extracting the tracer provider. - default_host (Optional[str]): Fallback transport host when ``api_endpoint`` is not + default_host (str | None): Fallback transport host when ``api_endpoint`` is not set on ``client_options``. Returns: @@ -982,8 +997,8 @@ def get_otel_interceptor( if not tracing_enabled and not error_logging_enabled: return None - custom_interceptors: list[Any] = [] - tracing_interceptor: Any = None + custom_interceptors: list[grpc.ClientInterceptor] = [] + tracing_interceptor: grpc.ClientInterceptor | None = None if tracing_enabled: import opentelemetry.instrumentation.grpc as otel_grpc # type: ignore[import-not-found] @@ -1027,9 +1042,9 @@ def get_otel_async_interceptor( """Returns async gRPC client interceptors for OpenTelemetry tracing and/or error logging. Args: - client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options + client_options (ClientOptions | dict[str, Any] | None): The client options object or dictionary used for feature gating and extracting the tracer provider. - default_host (Optional[str]): Fallback transport host when ``api_endpoint`` is not + default_host (str | None): Fallback transport host when ``api_endpoint`` is not set on ``client_options``. Returns: @@ -1041,7 +1056,7 @@ def get_otel_async_interceptor( if not tracing_enabled and not error_logging_enabled: return None - interceptors: list[Any] = [] + interceptors: list[grpc.aio.ClientInterceptor] = [] if tracing_enabled: # Ignored by mypy: Optional dependency only loaded if tracing_enabled is True import opentelemetry.instrumentation.grpc as otel_grpc # type: ignore[import-not-found] @@ -1084,10 +1099,10 @@ def apply_otel_async_channel_interceptors( Args: channel: The async gRPC channel to intercept. - client_options (Optional[Union[ClientOptions, dict[str, Any], Sequence[Any]]]): The client + client_options (ClientOptions | dict[str, Any] | Sequence[Any] | None): The client options used to resolve feature flags and tracer provider, or a sequence of pre-resolved interceptors. - default_host (Optional[str]): Fallback transport host when ``api_endpoint`` is not + default_host (str | None): Fallback transport host when ``api_endpoint`` is not set on ``client_options``. Returns: @@ -1299,9 +1314,9 @@ def _log_http_attempt( """Emits a DEBUG transport attempt log for a failed HTTP request (fail-open). Args: - status_code (Optional[Union[int, str]]): HTTP response status code or error code. - exc (Optional[BaseException]): Exception raised during HTTP dispatch, if any. - status_message (Optional[str]): Extracted HTTP error message fallback. + status_code (int | str | None): HTTP response status code or error code. + exc (BaseException | None): Exception raised during HTTP dispatch, if any. + status_message (str | None): Extracted HTTP error message fallback. """ try: resolved_url = ( @@ -1407,7 +1422,7 @@ def record_error(self, exc: BaseException | None) -> None: - ``status.message``: Error description message. Args: - exc (Optional[BaseException]): The exception raised during dispatch. + exc (BaseException | None): The exception raised during dispatch. """ if exc is None: return @@ -1537,13 +1552,13 @@ def trace_http_request( of instantiating a throwaway dummy request object on every single RPC execution. Args: - request (Optional[Any]): Bundled HTTP request object with .method, .url, .headers. - method (Optional[str]): Explicit HTTP method (e.g. "GET", "POST"). - url (Optional[str]): Fully qualified request URL. - url_template (Optional[str]): Parameterized path template for url.template. - headers (Optional[dict[str, Any]]): Mutable dictionary of request headers. - body (Optional[Any]): Request payload (used for body size calculation). - client_options (Optional[Union[ClientOptions, dict[str, Any]]]): Client options + request (Any | None): Bundled HTTP request object with .method, .url, .headers. + method (str | None): Explicit HTTP method (e.g. "GET", "POST"). + url (str | None): Fully qualified request URL. + url_template (str | None): Parameterized path template for url.template. + headers (dict[str, Any] | None): Mutable dictionary of request headers. + body (Any | None): Request payload (used for body size calculation). + client_options (ClientOptions | dict[str, Any] | None): Client options containing custom endpoint and tracer provider configurations. Returns: diff --git a/packages/google-api-core/google/api_core/client_logging.py b/packages/google-api-core/google/api_core/client_logging.py index 68d1f8d18398..e0bb49f02a61 100644 --- a/packages/google-api-core/google/api_core/client_logging.py +++ b/packages/google-api-core/google/api_core/client_logging.py @@ -1,17 +1,25 @@ +from __future__ import annotations + import json import logging import os from collections.abc import Mapping -from typing import Any, List, Optional +from typing import TYPE_CHECKING, Any, List, Optional from google.api_core import _feature_gating_helpers +if TYPE_CHECKING: + # flake8: ClientOptions is imported only for static analysis and type annotations + from google.api_core.client_options import ClientOptions # noqa: F401 + _LOGGING_INITIALIZED = False _BASE_LOGGER_NAME = "google" _LOGGER_PROVIDER = "logger_provider" -def is_error_logging_enabled(client_options: Any = None) -> bool: +def is_error_logging_enabled( + client_options: ClientOptions | dict[str, Any] | None = None, +) -> bool: """Determines if OpenTelemetry-correlated Production Error Logging (PEL) is enabled. This feature gate controls action-oriented error and retry logging @@ -20,7 +28,8 @@ def is_error_logging_enabled(client_options: Any = None) -> bool: Debug Logging (APDL) configured via ``GOOGLE_SDK_PYTHON_LOGGING_SCOPE``. Args: - client_options (Optional[Any]): The client options object or dictionary. + client_options (ClientOptions | dict[str, Any] | None): The client options + object or dictionary. Returns: bool: True if production error logging is enabled, False otherwise. diff --git a/packages/google-api-core/google/api_core/gapic_v1/method.py b/packages/google-api-core/google/api_core/gapic_v1/method.py index 5f30369d1e19..1f6bc072dba9 100644 --- a/packages/google-api-core/google/api_core/gapic_v1/method.py +++ b/packages/google-api-core/google/api_core/gapic_v1/method.py @@ -18,12 +18,15 @@ compression, pagination, and long-running operations to gRPC methods. """ +from __future__ import annotations + import asyncio import contextlib import enum import functools import logging -from typing import Any, List, Optional, Tuple +from collections.abc import Iterator +from typing import TYPE_CHECKING, Any, List, Tuple from google.api_core import ( _observability, @@ -34,6 +37,10 @@ from google.api_core.gapic_v1.client_info import METRICS_METADATA_KEY from google.api_core.timeout import TimeToDeadlineTimeout +if TYPE_CHECKING: + # flake8: opentelemetry.trace is imported only for static analysis and type annotations + import opentelemetry.trace # noqa: F401 + USE_DEFAULT_METADATA = object() @@ -266,19 +273,19 @@ def _prepare_call(self, timeout, retry, compression, kwargs): def _emit_method_error_log( self, exc: BaseException, - span: Any, + span: opentelemetry.trace.Span | None, error_attrs: dict[str, Any], is_cancelled: bool, - logger: Optional[logging.Logger] = None, + logger: logging.Logger | None = None, ) -> None: """Constructs and emits the L3 terminal method error/cancellation log record. Args: exc (BaseException): The terminal exception raised by the RPC method. - span (Any): The active Tier 3 OpenTelemetry span, if any. + span (opentelemetry.trace.Span | None): The active Tier 3 OpenTelemetry span, if any. error_attrs (dict[str, Any]): Pre-extracted telemetry error attributes. is_cancelled (bool): True if the RPC was cancelled rather than failed. - logger (Optional[logging.Logger]): Explicit logger override for testing. + logger (logging.Logger | None): Explicit logger override for testing. """ try: rpc_system = _observability._resolve_rpc_system(self._kind) @@ -314,15 +321,15 @@ def _emit_method_error_log( def _report_method_error( self, exc: BaseException, - span: Any = None, - logger: Optional[logging.Logger] = None, + span: opentelemetry.trace.Span | None = None, + logger: logging.Logger | None = None, ) -> None: """Enriches the active Tier 3 method span and emits an L3 terminal method error log. Args: exc (BaseException): The terminal exception raised by the RPC method. - span (Optional[Any]): The active Tier 3 OpenTelemetry span, if any. - logger (Optional[logging.Logger]): Explicit logger override for testing. + span (opentelemetry.trace.Span | None): The active Tier 3 OpenTelemetry span, if any. + logger (logging.Logger | None): Explicit logger override for testing. """ error_logging_enabled = client_logging.is_error_logging_enabled( self._client_options @@ -356,7 +363,7 @@ def _report_method_error( ) @contextlib.contextmanager - def _trace_span(self): + def _trace_span(self) -> Iterator[None]: """Context manager that wraps method execution in a Tier 3 OpenTelemetry span.""" span_cm = contextlib.nullcontext() if self._start_span_fn is not None: From 2672735581ae9028b9b6beffecdd034aaf8e3c21 Mon Sep 17 00:00:00 2001 From: chalmer lowe Date: Thu, 5 Nov 2026 17:03:30 -0500 Subject: [PATCH 4/4] refactor(api-core): simplify guards and document fail-open exception handlers --- .../google/api_core/_observability.py | 58 ++++++++++--------- .../google/api_core/gapic_v1/method.py | 14 +++-- .../tests/unit/gapic/test_method.py | 2 +- 3 files changed, 42 insertions(+), 32 deletions(-) diff --git a/packages/google-api-core/google/api_core/_observability.py b/packages/google-api-core/google/api_core/_observability.py index cb675df84c3c..1997a3d469cf 100644 --- a/packages/google-api-core/google/api_core/_observability.py +++ b/packages/google-api-core/google/api_core/_observability.py @@ -305,7 +305,7 @@ def _decode_method_name(method: str | bytes | None) -> str: """ if isinstance(method, bytes): return method.decode("utf-8", errors="replace") - return str(method or "") + return method or "" def extract_rpc_identity( @@ -334,7 +334,7 @@ def _resolve_rpc_system(kind: str | None = "grpc") -> str: Returns: str: "http" for REST/HTTP transports, or "grpc" otherwise. """ - if isinstance(kind, str) and kind.startswith(("rest", "http")): + if kind and kind.startswith(("rest", "http")): return "http" return "grpc" @@ -345,12 +345,12 @@ def _get_service_logger( ) -> logging.Logger: """Resolves a service-scoped logger matching GOOGLE_SDK_PYTHON_LOGGING_SCOPE. - 1. If `target` has a `__module__` starting with `google.` (excluding `google.api_core`), - returns `logging.getLogger(target.__module__)`. - 2. Otherwise, if `method_name` is a full gRPC method path (e.g. - `/google.cloud.secretmanager.v1.SecretManagerService/GetSecret`), extracts - the service package prefix (`google.cloud.secretmanager.v1`). - 3. Falls back to `logging.getLogger("google.api_core._observability")`. + - If `target` has a `__module__` starting with `google.` (excluding `google.api_core`), + returns `logging.getLogger(target.__module__)`. + - Otherwise, if `method_name` is a full gRPC method path (e.g. + `/google.cloud.secretmanager.v1.SecretManagerService/GetSecret`), extracts + the service package prefix (`google.cloud.secretmanager.v1`). + - Falls back to `logging.getLogger("google.api_core._observability")`. Args: method_name (str | bytes | None): Full gRPC method path. @@ -395,7 +395,7 @@ def _is_failed_attempt( return True if norm_status is None or norm_status == "OK": return False - return not (norm_status.isdigit() and int(norm_status) < 400) + return not (norm_status.isdecimal() and int(norm_status) < 400) def _build_attempt_log_extra( @@ -445,14 +445,16 @@ def _build_attempt_log_extra( "status": ( int(status_code) if isinstance(status_code, int) - or (isinstance(status_code, str) and status_code.isdigit()) + or (isinstance(status_code, str) and status_code.isdecimal()) else status_code ), } else: extra_payload["rpc.method"] = target - extra_payload["json_fields"] = dict(extra_payload) + # Take a shallow copy before assigning json_fields so json_fields does not + # reference extra_payload itself and cause a circular reference in json.dumps. + extra_payload["json_fields"] = extra_payload.copy() return extra_payload, resolved_status @@ -1339,6 +1341,8 @@ def _log_http_attempt( client_options=self._client_options, status_message=status_message, ) + # Intentionally catch broad Exception so telemetry logging fails open + # and never disrupts the caller's RPC flow. except Exception: pass @@ -1361,6 +1365,8 @@ def record_response(self, response: Any) -> None: self._log_http_attempt( status_code=code_int, status_message=error_msg ) + # Intentionally catch broad Exception so telemetry extraction/logging + # fails open and never disrupts the caller's RPC flow. except Exception: pass @@ -1394,13 +1400,10 @@ def record_response(self, response: Any) -> None: ) except (ValueError, TypeError): pass - elif hasattr(response, "_content") and response._content is not None: - try: - span.set_attribute( - "http.response.body.size", len(response._content) - ) - except Exception: - pass + elif getattr(response, "_content", None) is not None: + span.set_attribute("http.response.body.size", len(response._content)) + # Intentionally catch broad Exception so telemetry span enrichment + # fails open and never disrupts the caller's RPC flow. except Exception: pass @@ -1435,15 +1438,16 @@ def record_error(self, exc: BaseException | None) -> None: return try: - if hasattr(span, "set_attribute"): - if status_code: - span.set_attribute("error.type", str(status_code)) - else: - span.set_attribute("error.type", exc.__class__.__name__) - msg = str(exc) - if msg: - span.set_attribute("status.message", msg) - except Exception: # Fail-open on error attribute extraction failure + if status_code: + span.set_attribute("error.type", str(status_code)) + else: + span.set_attribute("error.type", exc.__class__.__name__) + msg = str(exc) + if msg: + span.set_attribute("status.message", msg) + # Intentionally catch broad Exception so telemetry span enrichment + # fails open and never disrupts the caller's RPC flow. + except Exception: pass # Alias for method name parity with Daniel's suggested convention diff --git a/packages/google-api-core/google/api_core/gapic_v1/method.py b/packages/google-api-core/google/api_core/gapic_v1/method.py index 1f6bc072dba9..3320f15c1461 100644 --- a/packages/google-api-core/google/api_core/gapic_v1/method.py +++ b/packages/google-api-core/google/api_core/gapic_v1/method.py @@ -297,7 +297,9 @@ def _emit_method_error_log( **error_attrs, **_observability.extract_trace_context(span), } - extra_payload["json_fields"] = dict(extra_payload) + # Take a shallow copy before assigning json_fields so json_fields does not + # reference extra_payload itself and cause a circular reference in json.dumps. + extra_payload["json_fields"] = extra_payload.copy() log_target = logger or _observability._get_service_logger( self._method_name, self._target @@ -315,6 +317,8 @@ def _emit_method_error_log( exc_info=not is_cancelled, extra=extra_payload, ) + # Intentionally catch broad Exception so telemetry logging fails open + # and never disrupts the caller's RPC flow. except Exception: pass @@ -347,16 +351,18 @@ def _report_method_error( if is_cancelled: error_attrs["rpc.response.status_code"] = "CANCELLED" - # 1. Enrich active Tier 3 method span - if span is not None and hasattr(span, "set_attribute"): + # Enrich active Tier 3 method span + if span is not None: try: for k, v in error_attrs.items(): if not k.startswith("exception."): span.set_attribute(k, v) + # Intentionally catch broad Exception so telemetry span enrichment + # fails open and never disrupts the caller's RPC flow. except Exception: pass - # 2. Emit L2/L3 Terminal Method Error Log + # Emit L2/L3 Terminal Method Error Log if error_logging_enabled: self._emit_method_error_log( exc, span, error_attrs, is_cancelled, logger=logger diff --git a/packages/google-api-core/tests/unit/gapic/test_method.py b/packages/google-api-core/tests/unit/gapic/test_method.py index 3e0402dcbd41..966391b30fca 100644 --- a/packages/google-api-core/tests/unit/gapic/test_method.py +++ b/packages/google-api-core/tests/unit/gapic/test_method.py @@ -917,7 +917,7 @@ def test_gapic_callable_report_method_error_and_logging(monkeypatch): cancel_logger = mock.Mock() wrapped_fail._report_method_error( - exceptions.Cancelled("Call cancelled"), span=mock_span, logger=cancel_logger + exceptions.Cancelled("Call cancelled"), span=None, logger=cancel_logger ) cancel_logger.log.assert_called_once() c_args, c_kwargs = cancel_logger.log.call_args