diff --git a/packages/google-api-core/google/api_core/_observability.py b/packages/google-api-core/google/api_core/_observability.py index e487687f5673..1997a3d469cf 100644 --- a/packages/google-api-core/google/api_core/_observability.py +++ b/packages/google-api-core/google/api_core/_observability.py @@ -19,12 +19,15 @@ 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, exceptions +from google.api_core import _feature_gating_helpers, client_logging, exceptions from google.api_core.client_options import ClientOptions try: @@ -65,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. @@ -97,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. @@ -215,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. @@ -288,6 +291,238 @@ 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 (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). + """ + 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 (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, + 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 (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 (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). + """ + 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.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, + 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, @@ -295,9 +530,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 +584,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 +647,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 +671,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,90 +819,262 @@ 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 (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 + + @contextlib.contextmanager + 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 + 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: 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_str = _decode_method_name(getattr(client_call_details, "method", None)) + with self._capture_attempt_exception(method_str): + response = continuation(client_call_details, request) + + 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 + + +class _AsyncGrpcAttemptLoggingInterceptor( + _BaseGrpcAttemptLoggingInterceptor, _AsyncUnaryUnaryClientInterceptor +): + """AsyncIO gRPC client interceptor that logs failed wire attempts at DEBUG level.""" + + __slots__ = () + + 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_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() + 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, + ) + 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( + 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 (ClientOptions | dict[str, Any] | None): Client options + used to resolve endpoint attributes and custom tracer provider. + default_host (str | None): 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 + 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) + 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 otel_interceptor(channel: grpc.Channel) -> grpc.Channel: - if grpc is not None: + 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. + + 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 + 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] + 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] - 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 @@ -694,10 +1101,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: @@ -900,12 +1307,69 @@ 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). """ + 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 + ) + # Intentionally catch broad Exception so telemetry extraction/logging + # fails open and never disrupts the caller's RPC flow. + except Exception: + pass + span = self._span if span is None or not hasattr(span, "set_attribute"): return @@ -916,11 +1380,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 +1389,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 +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 @@ -955,7 +1411,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 +1425,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 +1520,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 @@ -1094,13 +1556,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..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,11 +1,45 @@ +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" + + +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, + ) + # 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..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 @@ -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._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 +215,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 +270,107 @@ 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: + 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), + } + # 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): + 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: try: @@ -288,11 +387,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..966391b30fca 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=None, 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..74d488f3d70e 100644 --- a/packages/google-api-core/tests/unit/test_observability.py +++ b/packages/google-api-core/tests/unit/test_observability.py @@ -2338,3 +2338,435 @@ 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": "/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, 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" + + 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): + 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" + + 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") + + 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