diff --git a/packages/google-api-core/google/api_core/_observability.py b/packages/google-api-core/google/api_core/_observability.py index e487687f5673..819924407081 100644 --- a/packages/google-api-core/google/api_core/_observability.py +++ b/packages/google-api-core/google/api_core/_observability.py @@ -18,13 +18,17 @@ from __future__ import annotations +import asyncio import contextlib +import enum import functools +import logging import traceback import urllib.parse +from collections.abc import Iterator, MutableMapping 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: @@ -35,14 +39,23 @@ _suppress_instrumentation = contextlib.nullcontext if TYPE_CHECKING: - # flake8: grpc, trace, and ClientInterceptor are imported only for static analysis and type annotations + # flake8: grpc and opentelemetry.trace are imported only for static analysis and type annotations # The 'noqa: F401' comment avoids flake8 "imported but not used" errors. import grpc # noqa: F401 import opentelemetry.trace # noqa: F401 - from google.api_core.grpc_helpers import ClientInterceptor # noqa: F401 - +_LOGGER = logging.getLogger(__name__) _TRACER_PROVIDER = "tracer_provider" +_TRACE_CONTEXT_PROPAGATOR: Any = None +_CHANNEL_INTERCEPTOR_MAPPING = ( + ("intercept_unary_unary", "_unary_unary_interceptors"), + ("intercept_unary_stream", "_unary_stream_interceptors"), + ("intercept_stream_unary", "_stream_unary_interceptors"), + ("intercept_stream_stream", "_stream_stream_interceptors"), +) +_STANDARD_INTERCEPTOR_METHODS = tuple( + method for method, _ in _CHANNEL_INTERCEPTOR_MAPPING +) @functools.cache @@ -65,7 +78,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. @@ -97,7 +110,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. @@ -175,6 +188,28 @@ def _extract_error_info_fields(target_exc: Any) -> tuple[Any, Any, Any]: ) +def _extract_http_status_code(response: Any) -> int | None: + """Extracts the numeric HTTP status code from an HTTP response object. + + Inspects ``response.status_code`` (used by ``requests.Response``) and falls + back to ``response.status`` (used by ``urllib3`` and ``aiohttp``), returning + the status code only when it is a genuine integer (excluding ``bool``). + + Args: + response (Any): An HTTP response object (or ``GoogleAPICallError.response``). + + Returns: + int | None: The integer HTTP status code (e.g. 200, 404, 503), or None + if no integer status code is present. + """ + status = getattr(response, "status_code", None) + if status is None: + status = getattr(response, "status", None) + if isinstance(status, int) and not isinstance(status, bool): + return status + return None + + def _extract_error_message( target_exc: Any, http_response: Any = None, fallback: str | None = None ) -> str | None: @@ -215,11 +250,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. @@ -245,16 +280,15 @@ def build_telemetry_error_attributes( domain, reason, metadata = _extract_error_info_fields(target_exc) # GoogleAPICallError.response holds either an HTTP Response or a gRPC Call - # (which has trailing_metadata). Validate once up front so http_response is - # only retained when it is an actual HTTP response with an integer status code. + # (which has trailing_metadata). Retain http_response only when it is an + # HTTP response with a valid integer HTTP status code. http_response = getattr(target_exc, "response", None) http_status = ( - getattr(http_response, "status_code", getattr(http_response, "status", None)) - if http_response is not None and not hasattr(http_response, "trailing_metadata") + _extract_http_status_code(http_response) + if not hasattr(http_response, "trailing_metadata") else None ) - if not isinstance(http_status, int) or isinstance(http_status, bool): - http_status = None + if http_status is None: http_response = None # Populate OpenTelemetry semantic error attributes and ErrorInfo metadata @@ -288,6 +322,234 @@ def build_telemetry_error_attributes( return attrs +def _decode_method_name(method: str | bytes | None) -> str: + """Normalizes a string or bytes RPC method name to a UTF-8 string. + + Args: + method (str | bytes | None): 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 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 (str | bytes | None): 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). + """ + full_rpc_name = _decode_method_name(method_name).lstrip("/") + service, _, method = full_rpc_name.rpartition("/") + return full_rpc_name, 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 (str | None): The transport kind identifier. + + Returns: + str: "http" for REST/HTTP transports, or "grpc" otherwise. + """ + if kind and kind.startswith(("rest", "http")): + return "http" + return "grpc" + + +def _get_service_logger( + method_name: str | bytes | None = None, + target: Callable[..., Any] | None = None, +) -> logging.Logger: + """Resolves a service-scoped logger matching GOOGLE_SDK_PYTHON_LOGGING_SCOPE. + + - 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. + target (Callable[..., Any] | None): 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 (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. + """ + if exc is not None: + return True + if norm_status is None or norm_status == "OK": + return False + return not (norm_status.isdecimal() and int(norm_status) < 400) + + +def _build_attempt_log_extra( + span: opentelemetry.trace.Span | None, + target: str | None, + status_code: int | str | enum.Enum | None, + norm_status: str | None, + rpc_system: 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 (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. + rpc_system (str): Normalized OTel RPC system identifier ("grpc" or "http"). + 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], str | None]: A 2-tuple of (extra_payload, resolved_status). + """ + 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.isdecimal()) + else status_code + ), + } + else: + extra_payload["rpc.method"] = target + + # 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 + + +def log_transport_attempt( + span: opentelemetry.trace.Span | None = None, + target: 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: ClientOptions | dict[str, Any] | None = 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 (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 (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 + + 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 + + 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, + rpc_system=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, @@ -295,9 +557,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: @@ -349,7 +611,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: @@ -412,7 +674,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)(): @@ -436,7 +698,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: @@ -584,102 +846,315 @@ async def intercept_stream_stream( return await continuation(client_call_details, request_iterator) +@contextlib.contextmanager +def _capture_grpc_attempt_exception( + method_name: str, + client_options: ClientOptions | dict[str, Any] | None = None, +) -> Iterator[None]: + """Logs any unhandled exception raised during gRPC continuation dispatch and re-raises. + + Args: + method_name (str): Decoded gRPC method name. + client_options (ClientOptions | dict[str, Any] | None): Client options + used for error logging gating. + + Yields: + None + """ + try: + yield + except BaseException as exc: + log_transport_attempt( + target=method_name, + kind="grpc", + exc=exc, + client_options=client_options, + ) + raise + + +class _GrpcAttemptLoggingInterceptor(_SyncUnaryUnaryClientInterceptor): + """Sync gRPC client interceptor that logs failed wire attempts at DEBUG level. + + Args: + client_options (ClientOptions | dict[str, Any] | None): Client options + used for error logging gating. + """ + + __slots__ = ("_client_options",) + + def __init__( + self, client_options: ClientOptions | dict[str, Any] | None = None + ) -> None: + self._client_options = client_options + + def intercept_unary_unary( + 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[..., 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: + Any: The gRPC Call/Future response returned by ``continuation``. + """ + method_name = _decode_method_name(getattr(client_call_details, "method", None)) + with _capture_grpc_attempt_exception(method_name, self._client_options): + response = continuation(client_call_details, request) + + def _inspect_sync_response(resp: Any) -> None: + try: + exc_fn = getattr(resp, "exception", None) + if callable(exc_fn) and (resp_exc := exc_fn()) is not None: + code_fn = getattr(resp, "code", None) + log_transport_attempt( + target=method_name, + status_code=code_fn() if callable(code_fn) else None, + 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 + + add_callback = getattr(response, "add_done_callback", None) + if callable(add_callback): + add_callback(_inspect_sync_response) + else: + _inspect_sync_response(response) + return response + + +class _AsyncGrpcAttemptLoggingInterceptor(_AsyncUnaryUnaryClientInterceptor): + """AsyncIO gRPC client interceptor that logs failed wire attempts at DEBUG level. + + Args: + client_options (ClientOptions | dict[str, Any] | None): Client options + used for error logging gating. + """ + + __slots__ = ("_client_options",) + + def __init__( + self, client_options: ClientOptions | dict[str, Any] | None = None + ) -> None: + self._client_options = client_options + + async def intercept_unary_unary( + 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[..., 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: + Any: The async gRPC call object returned by ``continuation``. + """ + method_name = _decode_method_name(getattr(client_call_details, "method", None)) + with _capture_grpc_attempt_exception(method_name, self._client_options): + call = await continuation(client_call_details, request) + + async def _inspect_async_call(target_call: Any) -> None: + try: + code_fn = getattr(target_call, "code", None) + if ( + callable(code_fn) + and getattr(code := await code_fn(), "name", str(code)) != "OK" + ): + details_fn = getattr(target_call, "details", None) + details = await details_fn() if callable(details_fn) else "" + log_transport_attempt( + target=method_name, + status_code=code, + kind="grpc", + exc=Exception(details) if details else None, + client_options=self._client_options, + status_message=details or None, + ) + except Exception: + # Fail-open: never disrupt the caller's RPC flow on post-dispatch inspection errors + pass + + add_callback = getattr(call, "add_done_callback", None) + if callable(add_callback): + add_callback( + lambda finished_call: asyncio.ensure_future( + _inspect_async_call(finished_call) + ) + ) + else: + await _inspect_async_call(call) + 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 the keyword arguments passed to OpenTelemetry's gRPC interceptor factories. + + Both the sync (`otel_grpc.client_interceptor`) and async + (`otel_grpc.aio_client_interceptors`) factories in `opentelemetry.instrumentation.grpc` + accept the same keyword arguments to configure how gRPC client spans are created + and enriched with Google Cloud attributes. + + Args: + client_options (ClientOptions | dict[str, Any] | None): Client options + used to resolve endpoint attributes and a custom tracer provider. + default_host (str | None): Fallback transport host when ``api_endpoint`` is not + set on ``client_options``. + + Returns: + dict[str, Any]: A dictionary with ``tracer_provider``, ``request_hook``, and + ``response_hook`` ready to unpack (``**``) into the OpenTelemetry gRPC + interceptor factory functions. + """ + 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 + 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: - 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[grpc.ClientInterceptor] = [] + tracing_interceptor: grpc.ClientInterceptor | None = None - endpoint_attrs = _extract_endpoint_attributes( - client_options, default_host=default_host - ) - request_hook = _make_grpc_client_request_hook(endpoint_attrs) + # Order matters (outer -> inner): + # 1. tracing_interceptor (outermost: starts the Tier 4 gRPC span) + # 2. _GrpcAttemptLoggingInterceptor (middle: logs failed attempts with the + # Tier 4 span active, outside of suppress_instrumentation) + # 3. _SuppressingClientInterceptor (innermost: suppresses downstream + # generic auto-instrumentation without suppressing our attempt logs) + if error_logging_enabled: + custom_interceptors.append(_GrpcAttemptLoggingInterceptor(client_options)) - 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() + if tracing_enabled: + import opentelemetry.instrumentation.grpc as otel_grpc # type: ignore[import-not-found] + + tracing_interceptor = otel_grpc.client_interceptor( + **_build_otel_grpc_hook_kwargs(client_options, default_host) + ) + custom_interceptors.append(_SuppressingClientInterceptor()) + + 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. + """ + # Step 1 (inner layers): Wrap channel with custom_interceptors + # ([logging, suppressing]), where logging executes outside suppressing. + 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) + # Step 2 (outermost layer): Wrap tracing_interceptor around the outside + # of the channel, yielding: [tracing -> logging -> suppressing -> channel]. + 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 + 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: - 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] - - 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( - 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()) + # Order matters (outer -> inner): + # 1. tracing_interceptors (outermost: starts the Tier 4 async gRPC span) + # 2. _AsyncGrpcAttemptLoggingInterceptor (middle: logs failed attempts with + # the Tier 4 span active, outside of suppress_instrumentation) + # 3. _AsyncSuppressingClientInterceptor (innermost: suppresses downstream + # generic auto-instrumentation without suppressing our attempt logs) + interceptors: list[grpc.aio.ClientInterceptor] = [] + if error_logging_enabled: + interceptors.append(_AsyncGrpcAttemptLoggingInterceptor(client_options)) + + 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] + + raw_interceptors = otel_grpc.aio_client_interceptors( + **_build_otel_grpc_hook_kwargs(client_options, default_host) + ) + tracing_interceptors = ( + raw_interceptors + if isinstance(raw_interceptors, (list, tuple)) + else [raw_interceptors] + ) + # Sandwich logging between outer tracing and inner suppression (outer -> inner): + # [*tracing_interceptors, *logging_interceptor, _AsyncSuppressingClientInterceptor] + interceptors = [ + *tracing_interceptors, + *interceptors, + _AsyncSuppressingClientInterceptor(), + ] return interceptors -_CHANNEL_INTERCEPTOR_MAPPING = ( - ("intercept_unary_unary", "_unary_unary_interceptors"), - ("intercept_unary_stream", "_unary_stream_interceptors"), - ("intercept_stream_unary", "_stream_unary_interceptors"), - ("intercept_stream_stream", "_stream_stream_interceptors"), -) - - def apply_otel_async_channel_interceptors( channel: Any, client_options: ClientOptions | dict[str, Any] | Sequence[Any] | None = None, @@ -694,10 +1169,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: @@ -737,11 +1212,11 @@ def apply_otel_async_channel_interceptors( # lacking standard method names directly to the unary-unary list. unary_list = getattr(channel, "_unary_unary_interceptors", None) if isinstance(unary_list, list): - standard_methods = tuple(method for method, _ in _CHANNEL_INTERCEPTOR_MAPPING) fallback_items = [ i for i in interceptor_list - if not any(hasattr(i, m) for m in standard_methods) and i not in unary_list + if not any(hasattr(i, m) for m in _STANDARD_INTERCEPTOR_METHODS) + and i not in unary_list ] if fallback_items: unary_list[:0] = fallback_items @@ -749,9 +1224,6 @@ def apply_otel_async_channel_interceptors( return channel -_TRACE_CONTEXT_PROPAGATOR: Any = None - - class _TraceContext: """Context manager for tracing an HTTP wire request with OpenTelemetry. @@ -790,7 +1262,7 @@ def _build_http_span_attributes( method: str | None = None, url: str | None = None, url_template: str | None = None, - headers: dict[str, Any] | None = None, + headers: MutableMapping[str, Any] | None = None, body: Any = None, client_options: ClientOptions | dict[str, Any] | None = None, ) -> tuple[str, dict[str, Any], Any]: @@ -885,7 +1357,7 @@ def __init__( method: str | None = None, url: str | None = None, url_template: str | None = None, - headers: dict[str, Any] | None = None, + headers: MutableMapping[str, Any] | None = None, body: Any = None, client_options: ClientOptions | dict[str, Any] | None = None, ): @@ -900,12 +1372,57 @@ 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 (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 = ( + 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, + ) + # Intentionally catch broad Exception so telemetry logging fails open + # and never disrupts the caller's RPC flow. + 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). """ + code_int = _extract_http_status_code(response) + error_msg: str | None = None + if code_int is not None and code_int >= 400: + error_msg = self._extract_http_error_message(response) + self._log_http_attempt(status_code=code_int, status_message=error_msg) + span = self._span if span is None or not hasattr(span, "set_attribute"): return @@ -916,11 +1433,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 +1442,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) @@ -941,13 +1453,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 @@ -955,7 +1464,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 @@ -969,25 +1478,29 @@ 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 + + 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: - 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 @@ -1060,10 +1573,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 @@ -1074,7 +1589,7 @@ def trace_http_request( method: str | None = None, url: str | None = None, url_template: str | None = None, - headers: dict[str, Any] | None = None, + headers: MutableMapping[str, Any] | None = None, body: Any = None, client_options: ClientOptions | dict[str, Any] | None = None, ) -> _TraceContext: @@ -1094,13 +1609,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 53d9f3a5c574..70abf646f722 100644 --- a/packages/google-api-core/google/api_core/client_logging.py +++ b/packages/google-api-core/google/api_core/client_logging.py @@ -1,11 +1,20 @@ +from __future__ import annotations + import json import logging import os from collections.abc import Mapping -from typing import 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" # Fields to be included in the StructuredLogFormatter. # @@ -49,6 +58,30 @@ _PREFIX_FIELDS = tuple(f[:-1] for f in _recognized_logging_fields if f.endswith("*")) +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 + (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 (ClientOptions | dict[str, Any] | None): 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, + ) + + def logger_configured(logger) -> bool: """Determines whether `logger` has non-default configuration 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..033a29abfe39 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,16 +18,29 @@ compression, pagination, and long-running operations to gRPC methods. """ +from __future__ import annotations + +import asyncio import contextlib import enum import functools -from typing import List, Tuple - -from google.api_core import _observability, grpc_helpers +import logging +from collections.abc import Iterator +from typing import TYPE_CHECKING, Any, List, 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 +if TYPE_CHECKING: + # flake8: opentelemetry.trace is imported only for static analysis and type annotations + import opentelemetry.trace # noqa: F401 + USE_DEFAULT_METADATA = object() @@ -110,22 +123,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 +178,9 @@ def __init__( self._retry = retry self._timeout = timeout self._compression = compression + self._client_options = client_options + self._method_name = method_name + self._rpc_system = _observability._resolve_rpc_system(kind) # Pre-extract the x-goog-api-client header from the initialized metadata. self._x_goog_api_client, remaining = _extract_metrics_header(metadata) @@ -205,22 +205,18 @@ def __init__( try: from opentelemetry import trace - tracer_provider = None - if isinstance(client_options, dict): - tracer_provider = client_options.get("tracer_provider") - elif client_options is not None: - tracer_provider = getattr(client_options, "tracer_provider", None) + tracer_provider = _observability._get_tracer_provider(client_options) if tracer_provider is not None: tracer = tracer_provider.get_tracer("google.api_core") 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": self._rpc_system, "rpc.method": span_name, **endpoint_attrs, } @@ -270,9 +266,109 @@ def _prepare_call(self, timeout, retry, compression, kwargs): return wrapped_func + def _emit_method_error_log( + self, + exc: BaseException, + span: opentelemetry.trace.Span | None, + error_attrs: dict[str, Any], + is_cancelled: bool, + 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 (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 (logging.Logger | None): Explicit logger override for testing. + """ + try: + if is_cancelled: + error_attrs.pop("exception.stacktrace", None) + extra_payload: dict[str, Any] = { + "rpcName": self._method_name, + "rpc.system.name": self._rpc_system, + **error_attrs, + **_observability.extract_trace_context(span), + } + # 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 + ) + 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, + ) + # Intentionally catch broad Exception so telemetry logging fails open + # and never disrupts the caller's RPC flow. + except Exception: + pass + + def _report_method_error( + self, + exc: BaseException, + 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 (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 + ) + 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" + + # 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 + + # 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): - span_cm = contextlib.nullcontext() + def _trace_span(self) -> Iterator[None]: + """Context manager that wraps method execution in a Tier 3 OpenTelemetry span.""" + span_cm: contextlib.AbstractContextManager[opentelemetry.trace.Span | None] = ( + contextlib.nullcontext() + ) if self._start_span_fn is not None: try: span_cm = self._start_span_fn() @@ -288,11 +384,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..0756aded08d6 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,57 @@ 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( + mock_otel, + mock_service_logger, + 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 + + 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 pytest.raises(type(err)): + await wrapped() + + mock_otel.span.set_attribute.assert_any_call( + "rpc.response.status_code", expected_status + ) + mock_service_logger.log.assert_called_once() + args, kwargs = mock_service_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/conftest.py b/packages/google-api-core/tests/conftest.py index ad186435fe17..2afc4d89947a 100644 --- a/packages/google-api-core/tests/conftest.py +++ b/packages/google-api-core/tests/conftest.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import logging import os import sys import types @@ -99,3 +100,25 @@ def mock_async_channel(): channel._stream_unary_interceptors = [] channel._stream_stream_interceptors = [] return channel + + +@pytest.fixture +def mock_logger(): + """Provides a strict autospecced logging.Logger instance.""" + return mock.create_autospec(logging.Logger, instance=True, spec_set=True) + + +@pytest.fixture +def mock_service_logger(monkeypatch, mock_logger): + """Enables experimental error logging and patches _observability._get_service_logger.""" + from google.api_core import _observability + + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_LOGGING_ENABLED", "true") + with mock.patch.object( + _observability, + "_get_service_logger", + return_value=mock_logger, + autospec=True, + spec_set=True, + ): + yield mock_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 292f05a73b50..117bba6f7aa4 100644 --- a/packages/google-api-core/tests/unit/gapic/test_method.py +++ b/packages/google-api-core/tests/unit/gapic/test_method.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import concurrent.futures import datetime import sys import types @@ -32,6 +33,12 @@ from google.api_core.gapic_v1 import client_info from tests.helpers import assert_uninstrumented_gapic_callable +_DEFAULT_SPAN_ATTRIBUTES = { + "rpc.system.name": "grpc", + "rpc.method": "google.cloud.secretmanager.v1.SecretManagerService/ListSecrets", + "url.domain": "googleapis.com", +} + def _utcnow_monotonic(): curr_value = datetime.datetime.min @@ -352,13 +359,6 @@ def test__deduplicate_metadata_tokens(headers, expected): assert dedup(*headers) == expected -_DEFAULT_SPAN_ATTRIBUTES = { - "rpc.system.name": "grpc", - "rpc.method": "google.cloud.secretmanager.v1.SecretManagerService/ListSecrets", - "url.domain": "googleapis.com", -} - - @pytest.mark.parametrize( "kwargs,capabilities_enabled", [ @@ -849,3 +849,112 @@ def test_wrap_method_uninstrumented_exception(): with pytest.raises(RuntimeError, match="uninstrumented error"): wrapped() + + +def test_gapic_callable_logging_silent_on_success(mock_service_logger): + """Proves _GapicCallable does not emit error logs when the RPC succeeds.""" + wrapped_ok = google.api_core.gapic_v1.method.wrap_method( + mock.Mock(return_value="ok"), + method_name="/google.cloud.secretmanager.v1.SecretManagerService/GetSecret", + ) + assert wrapped_ok() == "ok" + mock_service_logger.log.assert_not_called() + + +def test_gapic_callable_logging_terminal_error(mock_otel, mock_service_logger): + """Proves _GapicCallable enriches spans and emits an ERROR log with ErrorInfo on failure.""" + import logging + + 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 + mock_otel.span.get_span_context.return_value = types.SimpleNamespace( + is_valid=True, + trace_id=0x0123456789ABCDEF0123456789ABCDEF, + span_id=0x0123456789ABCDEF, + trace_flags=0x01, + ) + 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", + ) + + with pytest.raises(exceptions.NotFound): + wrapped_fail() + + mock_service_logger.log.assert_called_once() + args, kwargs = mock_service_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" + + +@pytest.mark.parametrize( + "cancel_exc_factory,expected_error_type", + [ + pytest.param( + lambda: exceptions.Cancelled("Call cancelled"), + "CANCELLED", + id="google_api_core_cancelled", + ), + pytest.param( + lambda: concurrent.futures.CancelledError("Future cancelled"), + "CancelledError", + id="concurrent_futures_cancelled_error", + ), + ], +) +def test_gapic_callable_logging_cancellation( + mock_otel, mock_service_logger, cancel_exc_factory, expected_error_type +): + """Proves _GapicCallable emits DEBUG without stacktrace on RPC or Future cancellation.""" + import logging + + err = cancel_exc_factory() + wrapped_cancel = google.api_core.gapic_v1.method.wrap_method( + mock.Mock(side_effect=err), + method_name="/google.cloud.secretmanager.v1.SecretManagerService/GetSecret", + ) + with pytest.raises(type(err)): + wrapped_cancel() + + mock_service_logger.log.assert_called_once() + c_args, c_kwargs = mock_service_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" + assert c_kwargs["extra"]["error.type"] == expected_error_type + + +def test_gapic_callable_logging_fail_open_resilience(mock_otel, mock_service_logger): + """Proves _GapicCallable fails open and re-raises caller exception if span or logger raises.""" + mock_otel.span.set_attribute.side_effect = RuntimeError("span broken") + mock_service_logger.log.side_effect = RuntimeError("logger broken") + + wrapped_fail = google.api_core.gapic_v1.method.wrap_method( + mock.Mock(side_effect=exceptions.NotFound("Secret missing")), + method_name="/google.cloud.secretmanager.v1.SecretManagerService/GetSecret", + ) + with pytest.raises(exceptions.NotFound, match="Secret missing"): + wrapped_fail() 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..00c460622c79 100644 --- a/packages/google-api-core/tests/unit/test_client_logging.py +++ b/packages/google-api-core/tests/unit/test_client_logging.py @@ -8,6 +8,10 @@ setup_logging, ) +DEFAULT_CONSUMER_PROJECT = "projects/123" +SAMPLE_TRACE_ID = "0123456789abcdef0123456789abcdef" +SAMPLE_SPAN_ID = "0123456789abcdef" + def reset_logger(scope): logger = logging.getLogger(scope) @@ -139,11 +143,6 @@ def test_structured_log_formatter(): assert parsed_msg["rpcName"] == "bar" -DEFAULT_CONSUMER_PROJECT = "projects/123" -SAMPLE_TRACE_ID = "0123456789abcdef0123456789abcdef" -SAMPLE_SPAN_ID = "0123456789abcdef" - - def test_structured_log_formatter_json_fields_and_wildcard_metadata(): record = logging.LogRecord( name="google.cloud.secretmanager_v1", @@ -223,3 +222,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..716f739a8d7b 100644 --- a/packages/google-api-core/tests/unit/test_observability.py +++ b/packages/google-api-core/tests/unit/test_observability.py @@ -26,6 +26,13 @@ from google.api_core._feature_gating_helpers import FeatureGatingError from google.api_core.client_options import ClientOptions +_INVALID_ARGUMENT_CODE = ( + "INVALID_ARGUMENT" if exceptions.grpc is not None else "BAD_REQUEST" +) +_UNAVAILABLE_CODE = ( + "UNAVAILABLE" if exceptions.grpc is not None else "SERVICE_UNAVAILABLE" +) + def test_is_otel_capabilities_enabled_disabled(monkeypatch): """Proves that is_otel_capabilities_enabled returns False when the tracing environment variable is disabled.""" @@ -2020,14 +2027,6 @@ def test_extract_error_message(target_exc, http_response, fallback, expected): ) -_INVALID_ARGUMENT_CODE = ( - "INVALID_ARGUMENT" if exceptions.grpc is not None else "BAD_REQUEST" -) -_UNAVAILABLE_CODE = ( - "UNAVAILABLE" if exceptions.grpc is not None else "SERVICE_UNAVAILABLE" -) - - @pytest.mark.parametrize( "exc,kwargs,expected", [ @@ -2338,3 +2337,581 @@ 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,expected_full,expected_service,expected_method", + [ + pytest.param(None, "", "", "", id="none_method_name"), + pytest.param("", "", "", "", id="empty_string_method_name"), + pytest.param( + b"/my.Service/Method", + "my.Service/Method", + "my.Service", + "Method", + id="bytes_method_name", + ), + pytest.param( + "/my.Service/Method", + "my.Service/Method", + "my.Service", + "Method", + id="string_method_name", + ), + ], +) +def test_extract_rpc_identity( + method_name, expected_full, expected_service, expected_method +): + """Proves extract_rpc_identity decodes and partitions method names safely.""" + full, service, method = _observability.extract_rpc_identity(method_name) + assert full == expected_full + assert service == expected_service + assert method == expected_method + + +@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": "/google.showcase.v1beta1.Echo/Echo", + "status_code": types.SimpleNamespace(name="OK"), + "kind": "grpc", + }, + id="grpc_enum_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, mock_logger, 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 + ) + _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", + ), + pytest.param( + { + "target": "/google.showcase.v1beta1.Echo/Echo", + "status_code": "UNKNOWN", + "kind": "grpc", + }, + { + "rpc.system.name": "grpc", + "rpc.method": "/google.showcase.v1beta1.Echo/Echo", + "rpc.response.status_code": "UNKNOWN", + "error.type": "UNKNOWN", + "status.message": "HTTP UNKNOWN", + }, + id="failed_grpc_attempt_without_exception", + ), + ], +) +def test_log_transport_attempt_emits_debug( + monkeypatch, mock_logger, 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(spec_set=["get_span_context", "set_attribute", "is_recording"]) + span.get_span_context.return_value = valid_ctx + + _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 _make_callback_sync_response(exc=None, code_name=None): + resp = mock.Mock(spec_set=["add_done_callback", "exception", "code"]) + resp.exception.return_value = exc + resp.code.return_value = ( + types.SimpleNamespace(name=code_name) if code_name else None + ) + resp.add_done_callback.side_effect = lambda cb: cb(resp) + return resp + + +@pytest.mark.parametrize( + "response_factory,expect_log,expected_code_name", + [ + pytest.param( + lambda: _make_callback_sync_response(exc=None), + False, + None, + id="ok_future_response_via_callback_no_log", + ), + pytest.param( + object, + False, + None, + id="plain_object_without_callback_no_log", + ), + pytest.param( + lambda: _make_callback_sync_response( + exc=RuntimeError("wire failure"), code_name="UNAVAILABLE" + ), + True, + "UNAVAILABLE", + id="failed_future_response_via_callback_logs", + ), + pytest.param( + lambda: types.SimpleNamespace( + exception=lambda: RuntimeError("wire failure without code") + ), + True, + None, + id="direct_response_no_code_logs_none", + ), + pytest.param( + lambda: mock.Mock( + spec_set=["exception"], + exception=mock.Mock(side_effect=RuntimeError("FutureCancelledError")), + ), + False, + None, + id="exception_in_inspection_fails_open", + ), + ], +) +def test_grpc_attempt_logging_interceptor_sync_responses( + monkeypatch, response_factory, expect_log, expected_code_name +): + """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") + response = response_factory() + + with mock.patch.object( + _observability, "log_transport_attempt", autospec=True, spec_set=True + ) as mock_log: + res = interceptor.intercept_unary_unary( + lambda d, r: response, call_details, "req" + ) + assert res is response + if expect_log: + mock_log.assert_called_once() + assert ( + mock_log.call_args[1]["target"] == "/google.showcase.v1beta1.Echo/Echo" + ) + if expected_code_name is not None: + assert mock_log.call_args[1]["status_code"].name == expected_code_name + else: + assert mock_log.call_args[1]["status_code"] is None + else: + mock_log.assert_not_called() + + +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", + [ + pytest.param(_FakeSyncRpcError, "INTERNAL", id="valid_status_code"), + pytest.param( + _BadCodeSyncRpcError, + "_BadCodeSyncRpcError", + id="status_code_method_raises", + ), + ], +) +def test_grpc_attempt_logging_interceptor_sync_exceptions( + mock_service_logger, exc_cls, expected_status +): + """Proves that _GrpcAttemptLoggingInterceptor logs exceptions raised directly by sync continuation.""" + interceptor = _observability._GrpcAttemptLoggingInterceptor() + call_details = types.SimpleNamespace(method=b"/google.showcase.v1beta1.Echo/Echo") + + def raising_continuation(d, r): + raise exc_cls("sync wire error") + + with pytest.raises(exc_cls): + interceptor.intercept_unary_unary(raising_continuation, call_details, "req") + mock_service_logger.debug.assert_called_once() + assert ( + mock_service_logger.debug.call_args[1]["extra"]["rpc.response.status_code"] + == expected_status + ) + + +def _make_callback_async_call(code_name="OK", details=""): + call_obj = mock.Mock(spec_set=["add_done_callback", "code", "details"]) + call_obj.code = mock.AsyncMock(return_value=types.SimpleNamespace(name=code_name)) + call_obj.details = mock.AsyncMock(return_value=details) + call_obj.add_done_callback = mock.Mock(side_effect=lambda cb: cb(call_obj)) + return call_obj + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "call_factory,expect_log,expected_code_name,expected_message", + [ + pytest.param( + lambda: _make_callback_async_call(code_name="OK"), + False, + None, + None, + id="ok_call_via_callback_no_log", + ), + pytest.param( + object, + False, + None, + None, + id="plain_object_without_callback_no_log", + ), + pytest.param( + lambda: _make_callback_async_call( + code_name="UNAVAILABLE", details="backend unavailable" + ), + True, + "UNAVAILABLE", + "backend unavailable", + id="failed_call_via_callback_logs", + ), + pytest.param( + lambda: mock.AsyncMock( + spec_set=["code", "details"], + code=mock.AsyncMock( + return_value=types.SimpleNamespace(name="UNAVAILABLE") + ), + details=mock.AsyncMock(return_value="direct await unavailable"), + ), + True, + "UNAVAILABLE", + "direct await unavailable", + id="direct_call_without_callback_logs", + ), + pytest.param( + lambda: mock.AsyncMock( + spec_set=["code"], + code=mock.AsyncMock(side_effect=RuntimeError("broken code inspection")), + ), + False, + None, + None, + id="broken_code_inspection_fails_open", + ), + ], +) +async def test_grpc_attempt_logging_interceptor_async( + monkeypatch, call_factory, expect_log, expected_code_name, expected_message +): + """Proves that _AsyncGrpcAttemptLoggingInterceptor logs failed async gRPC wire attempts.""" + import asyncio + + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_LOGGING_ENABLED", "true") + interceptor = _observability._AsyncGrpcAttemptLoggingInterceptor() + call_details = types.SimpleNamespace(method=b"/google.showcase.v1beta1.Echo/Echo") + call_obj = call_factory() + + with mock.patch.object( + _observability, "log_transport_attempt", autospec=True, spec_set=True + ) as mock_log: + res = await interceptor.intercept_unary_unary( + mock.AsyncMock(return_value=call_obj), call_details, "req" + ) + # Allow any scheduled done-callback task to run on the event loop + await asyncio.sleep(0) + assert res is call_obj + if expect_log: + 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"].name == expected_code_name + assert mock_log.call_args[1]["status_message"] == expected_message + else: + mock_log.assert_not_called() + + +@pytest.mark.asyncio +async def test_grpc_attempt_logging_interceptor_async_exceptions(mock_service_logger): + """Proves that _AsyncGrpcAttemptLoggingInterceptor logs exceptions raised directly by async continuation.""" + interceptor = _observability._AsyncGrpcAttemptLoggingInterceptor() + call_details = types.SimpleNamespace(method=b"/google.showcase.v1beta1.Echo/Echo") + + class FakeAsyncError(Exception): + def code(self): + raise ValueError("broken code") + + async def raising_continuation(d, r): + raise FakeAsyncError("async wire error") + + with pytest.raises(FakeAsyncError): + await interceptor.intercept_unary_unary( + raising_continuation, call_details, "req" + ) + mock_service_logger.debug.assert_called_once() + assert ( + mock_service_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 + if tracing_enabled == "true": + assert isinstance( + mock_ic.call_args[0][1], + _observability._GrpcAttemptLoggingInterceptor, + ) + assert isinstance( + mock_ic.call_args[0][2], + _observability._SuppressingClientInterceptor, + ) + 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 + ) + if tracing_enabled == "true": + assert isinstance( + async_list[1], _observability._AsyncGrpcAttemptLoggingInterceptor + ) + assert isinstance( + async_list[2], _observability._AsyncSuppressingClientInterceptor + ) + + +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