From 6fcf1d7bf9deedaba6b7f540f71549bd1e8c9821 Mon Sep 17 00:00:00 2001 From: chalmer lowe Date: Wed, 28 Oct 2026 12:16:32 -0400 Subject: [PATCH] feat(api_core): centralize telemetry error attribute and trace context builders in _observability - Move error attribute extraction from gapic_v1.method into _observability.build_telemetry_error_attributes with single-pass exception unwrapping, rpc.response.status_code resolution, native GoogleAPICallError property access, raw gRPC RpcError trailing_metadata parsing, and optional exception.* fields. - Update gapic_v1.method and tests to call _observability.build_telemetry_error_attributes directly without separate _extract_status_code calls. - Add has_opentelemetry() (with module-level caching) and extract_trace_context() in _observability. - Add unit tests achieving 100% statement and branch coverage. --- .../google/api_core/_observability.py | 231 +++++++++++++++++- .../google/api_core/gapic_v1/method.py | 103 +------- packages/google-api-core/tests/conftest.py | 1 + .../tests/unit/gapic/test_method.py | 85 ++++--- .../tests/unit/test_observability.py | 172 +++++++++++++ 5 files changed, 463 insertions(+), 129 deletions(-) diff --git a/packages/google-api-core/google/api_core/_observability.py b/packages/google-api-core/google/api_core/_observability.py index 34e1ca152c37..0442e07fb5e5 100644 --- a/packages/google-api-core/google/api_core/_observability.py +++ b/packages/google-api-core/google/api_core/_observability.py @@ -19,10 +19,11 @@ from __future__ import annotations import contextlib +import traceback import urllib.parse from typing import TYPE_CHECKING, Any, Callable, Sequence -from google.api_core import _feature_gating_helpers +from google.api_core import _feature_gating_helpers, exceptions from google.api_core.client_options import ClientOptions try: @@ -41,6 +42,22 @@ from google.api_core.grpc_helpers import ClientInterceptor # noqa: F401 _TRACER_PROVIDER = "tracer_provider" +_HAS_OPENTELEMETRY: bool | None = None + + +def has_opentelemetry() -> bool: + """Checks once per process if opentelemetry-api is installed.""" + global _HAS_OPENTELEMETRY + if _HAS_OPENTELEMETRY is None: + try: + # Ignored by mypy: Optional dependency only checked at runtime + # flake8: Imported only to verify package installation + import opentelemetry.trace # type: ignore[import-not-found] # noqa: F401 + + _HAS_OPENTELEMETRY = True + except ImportError: + _HAS_OPENTELEMETRY = False + return _HAS_OPENTELEMETRY def is_otel_capabilities_enabled( @@ -74,6 +91,218 @@ def is_otel_capabilities_enabled( return False +def extract_trace_context(span: Any = None) -> dict[str, Any]: + """Extracts trace_id, span_id, and Cloud Logging trace correlation fields. + + If `span` is None, falls back to `opentelemetry.trace.get_current_span()` + when OpenTelemetry is installed. Returns an empty dict if no valid recording + span context exists or if OpenTelemetry is not installed. + + Args: + span (Optional[Any]): The active OpenTelemetry span, if any. + + Returns: + dict[str, Any]: Trace and span correlation attributes. + """ + if span is None and has_opentelemetry(): + try: + # Ignored by mypy: Optional dependency only loaded after + # has_opentelemetry() check + from opentelemetry import trace # type: ignore[import-not-found] + + span = trace.get_current_span() + # Defensive fail-open: telemetry correlation must never crash an RPC call + # if a custom or broken OpenTelemetry TracerProvider raises at runtime. + except Exception: + span = None + + if span is None or not hasattr(span, "get_span_context"): + return {} + + try: + ctx = span.get_span_context() + # Defensive fail-open: guard against broken third-party or mock span objects. + except Exception: + return {} + + if ctx is None or not getattr(ctx, "is_valid", False): + return {} + + try: + trace_id_hex = f"{ctx.trace_id:032x}" + span_id_hex = f"{ctx.span_id:016x}" + trace_flags = getattr(ctx, "trace_flags", None) + if trace_flags is not None and hasattr(trace_flags, "sampled"): + sampled = bool(trace_flags.sampled) + elif isinstance(trace_flags, int): + sampled = bool(trace_flags & 0x01) + else: + sampled = False + except (TypeError, ValueError, AttributeError): + return {} + + return { + "trace_id": trace_id_hex, + "span_id": span_id_hex, + "logging.googleapis.com/trace": trace_id_hex, + "logging.googleapis.com/spanId": span_id_hex, + "logging.googleapis.com/trace_sampled": sampled, + } + + +def _extract_error_info_fields(target_exc: Any) -> tuple[Any, Any, Any]: + """Resolves (domain, reason, metadata) from an exception, ErrorInfo, or dict.""" + # Prefer native GoogleAPICallError properties (.domain, .reason, .metadata) + # when present; fall back to .error_info or raw gRPC trailing_metadata. + error_info = getattr(target_exc, "error_info", None) + if error_info is None and not ( + getattr(target_exc, "reason", None) + or getattr(target_exc, "domain", None) + or getattr(target_exc, "metadata", None) + ): + rpc_call = ( + target_exc + if hasattr(target_exc, "trailing_metadata") + else getattr(target_exc, "response", None) + ) + if rpc_call is not None and hasattr(rpc_call, "trailing_metadata"): + try: + _, error_info = exceptions._parse_grpc_error_details(rpc_call) + # Defensive fail-open: malformed trailing metadata must not abort + # error attribute extraction. + except Exception: + pass + + source = error_info if error_info is not None else target_exc + if isinstance(source, dict): + return ( + source.get("domain"), + source.get("reason"), + source.get("metadata"), + ) + return ( + getattr(source, "domain", None), + getattr(source, "reason", None), + getattr(source, "metadata", None), + ) + + +def _extract_error_message( + target_exc: Any, http_response: Any = None, fallback: str | None = None +) -> str | None: + """Resolves the human-readable error message across HTTP, gRPC, and Python errors.""" + http_msg = ( + _TraceContext._extract_http_error_message(http_response) + if http_response is not None + else None + ) + details = getattr(target_exc, "details", None) + details_msg = details() if callable(details) else details + exc_msg = str(target_exc) if isinstance(target_exc, BaseException) else None + raw_message = ( + http_msg + or getattr(target_exc, "message", None) + or details_msg + or fallback + or exc_msg + ) + return str(raw_message) if raw_message else None + + +def build_telemetry_error_attributes( + exc: BaseException | None, + status_code: str | None = None, + status_message: str | None = None, + include_exception: bool = False, +) -> dict[str, Any]: + """Extracts rpc.response.status_code, gcp.errors.*, error.type, status.message, and exception.* fields. + + Error details and ErrorInfo structures are resolved by inspecting: + * Chained exceptions: Unwraps RetryError or __cause__ to the root error. + * Canonical status code: Resolves rpc.response.status_code via + exceptions._extract_status_code. + * GoogleAPICallError & ErrorInfo attributes: Reads .domain, .reason, and + .metadata directly (or from .error_info / gRPC trailing_metadata). + * Unified attribute extraction: Populates gcp.errors.domain, error.type, + gcp.errors.metadata.*, status.message, and optional exception.* fields. + + Args: + exc (Optional[BaseException]): An exception (e.g. GoogleAPICallError + or grpc.RpcError) or ErrorInfo object. + status_code (Optional[str]): 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 + exc is None or lacks a message. + include_exception (bool): When True, also populates exception.type, + exception.message, and exception.stacktrace. + + Returns: + dict[str, Any]: Extracted error and exception attributes. + """ + if exc is None: + if status_code is None: + return {} + code_str = str(status_code) + return { + "rpc.response.status_code": code_str, + "error.type": code_str, + "status.message": status_message or f"HTTP {code_str}", + } + + # Unwrap chained exceptions once and resolve the canonical status code + target_exc = getattr(exc, "cause", None) or getattr(exc, "__cause__", None) or exc + resolved_code = exceptions._extract_status_code(target_exc, default=status_code) + domain, reason, metadata = _extract_error_info_fields(target_exc) + + # Identify HTTP transport responses (excluding gRPC Call objects) + 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") + else None + ) + + # Populate OpenTelemetry semantic error attributes and ErrorInfo metadata + attrs: dict[str, Any] = {"rpc.response.status_code": resolved_code} + if domain: + attrs["gcp.errors.domain"] = domain + if reason: + attrs["error.type"] = reason + elif ( + http_status is not None + and not isinstance(http_status, bool) + and str(http_status).isdigit() + ): + attrs["error.type"] = str(http_status) + else: + # Every failed span or log records a low-cardinality error.type; reuse + # the resolved canonical status code name or exception class name. + attrs["error.type"] = resolved_code + + if metadata and hasattr(metadata, "items"): + for k, v in metadata.items(): + attrs[f"gcp.errors.metadata.{k}"] = str(v) + + # Extract human-readable error description for cross-language parity + message = _extract_error_message( + target_exc, + http_response if http_status is not None else None, + status_message, + ) + if message: + attrs["status.message"] = message + + # Populate exception.* diagnostic fields for structured logs when requested + if include_exception and isinstance(exc, BaseException): + attrs["exception.type"] = exc.__class__.__name__ + attrs["exception.message"] = str(exc) + attrs["exception.stacktrace"] = "".join( + traceback.format_exception(type(exc), exc, exc.__traceback__) + ) + + return attrs + + def _extract_endpoint_attributes( client_options: ClientOptions | dict[str, Any] | None = None, default_host: str | None = None, 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 698a4fee9d4b..2dd3073da70d 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 @@ -21,9 +21,9 @@ import contextlib import enum import functools -from typing import Any, List, Optional, Tuple +from typing import List, Tuple -from google.api_core import _observability, exceptions, grpc_helpers +from google.api_core import _observability, 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 @@ -126,98 +126,6 @@ def _extract_rpc_identity( return method_str, service, method -def _extract_error_attributes(exc: Optional[BaseException]) -> dict[str, Any]: - """Extract gcp.errors.* and error.type attributes from an exception. - - Error details and ErrorInfo structures are resolved by inspecting the following locations: - * Chained exceptions: Unwraps RetryError or __cause__ to the root exception. - * GoogleAPICallError attributes: Reads ErrorInfo from ._error_info or .error_info. - * Native gRPC trailing metadata: Parses google.rpc.Status binary details from trailing_metadata. - * Unified attribute extraction: Extracts domain, reason, and metadata from ErrorInfo or exception attributes. - - Args: - exc (Optional[BaseException]): An exception (such as GoogleAPICallError or grpc.RpcError) or ErrorInfo object. - - Returns: - dict[str, Any]: Extracted error attributes (e.g. gcp.errors.domain, error.type, gcp.errors.metadata.*). - """ - attrs: dict[str, Any] = {} - if exc is None: - return attrs - - # 1. Unwrap chained exceptions: unwrap RetryError or __cause__ to the root failure - target_exc = getattr(exc, "cause", None) or getattr(exc, "__cause__", None) or exc - - # 2. Check GoogleAPICallError ErrorInfo attributes - error_info = getattr(target_exc, "_error_info", None) or getattr( - target_exc, "error_info", None - ) - - # 3. Check native gRPC trailing metadata for binary google.rpc.Status details - if error_info is None: - rpc_call = ( - target_exc - if hasattr(target_exc, "trailing_metadata") - else getattr(target_exc, "response", None) - ) - if rpc_call is not None and hasattr(rpc_call, "trailing_metadata"): - try: - _, error_info = exceptions._parse_grpc_error_details(rpc_call) - except Exception: - pass - - # 4. Unified attribute extraction: extract domain, reason, and metadata from ErrorInfo or exception attributes - 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") - else None - ) - - source = error_info or target_exc - if isinstance(source, dict): - domain = source.get("domain") - reason = source.get("reason") - metadata = source.get("metadata") - else: - domain = getattr(source, "domain", None) - reason = getattr(source, "reason", None) - metadata = getattr(source, "metadata", None) - - if domain: - attrs["gcp.errors.domain"] = domain - if reason: - attrs["error.type"] = reason - elif isinstance(http_status, int): - attrs["error.type"] = str(http_status) - else: - # Fallback per OpenTelemetry Semantic Conventions: every failed span should record - # a low-cardinality error.type. Use canonical status code name or exception class name. - status_code = exceptions._extract_status_code(target_exc) - attrs["error.type"] = status_code or target_exc.__class__.__name__ - if metadata: - for k, v in metadata.items(): - attrs[f"gcp.errors.metadata.{k}"] = str(v) - - # 5. Extract human-readable error description for cross-language PRD parity - message = ( - _observability._TraceContext._extract_http_error_message(http_response) - if http_status is not None - else None - ) - if not message: - message = getattr(target_exc, "message", None) - if not message and hasattr(target_exc, "details"): - details = target_exc.details - message = details() if callable(details) else details - if not message and isinstance(target_exc, BaseException): - message = str(target_exc) - if message: - attrs["status.message"] = str(message) - - return attrs - - class _GapicCallable(object): """Callable that applies retry, timeout, and metadata logic. @@ -381,10 +289,9 @@ def _trace_span(self): raise except BaseException as exc: if span is not None and hasattr(span, "set_attribute"): - span.set_attribute( - "rpc.response.status_code", exceptions._extract_status_code(exc) - ) - for k, v in _extract_error_attributes(exc).items(): + for k, v in _observability.build_telemetry_error_attributes( + exc + ).items(): span.set_attribute(k, v) raise diff --git a/packages/google-api-core/tests/conftest.py b/packages/google-api-core/tests/conftest.py index d8ef2918c9f2..ad186435fe17 100644 --- a/packages/google-api-core/tests/conftest.py +++ b/packages/google-api-core/tests/conftest.py @@ -43,6 +43,7 @@ def mock_otel(monkeypatch): mock_trace = mock.Mock() mock_trace.get_tracer.return_value = mock_tracer + mock_trace.get_current_span.return_value = mock_span mock_trace.SpanKind.CLIENT = "CLIENT" mock_trace.SpanKind.INTERNAL = "INTERNAL" mock_trace.StatusCode.ERROR = "ERROR" 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 ad094fbe46dd..929e80f269ca 100644 --- a/packages/google-api-core/tests/unit/gapic/test_method.py +++ b/packages/google-api-core/tests/unit/gapic/test_method.py @@ -26,8 +26,8 @@ import google.api_core.gapic_v1.method import google.api_core.page_iterator +from google.api_core import _observability, exceptions, retry, timeout from google.api_core import client_options as client_options_lib -from google.api_core import exceptions, retry, timeout from google.api_core.gapic_v1 import client_info from tests.helpers import assert_uninstrumented_gapic_callable @@ -781,18 +781,24 @@ def test_wrap_method_otel_tracing_endpoint_attributes_from_default_host( def test_extract_error_attributes_standard_exception(): - """Proves that _extract_error_attributes returns fallback error.type for exceptions without ErrorInfo.""" - assert google.api_core.gapic_v1.method._extract_error_attributes( - ValueError("fail") - ) == {"error.type": "ValueError", "status.message": "fail"} - assert google.api_core.gapic_v1.method._extract_error_attributes( + """Proves that build_telemetry_error_attributes returns fallback error.type for exceptions without ErrorInfo.""" + assert _observability.build_telemetry_error_attributes(ValueError("fail")) == { + "rpc.response.status_code": "ValueError", + "error.type": "ValueError", + "status.message": "fail", + } + assert _observability.build_telemetry_error_attributes( exceptions.InvalidArgument("invalid argument") - ) == {"error.type": "INVALID_ARGUMENT", "status.message": "invalid argument"} - assert google.api_core.gapic_v1.method._extract_error_attributes(None) == {} + ) == { + "rpc.response.status_code": "INVALID_ARGUMENT", + "error.type": "INVALID_ARGUMENT", + "status.message": "invalid argument", + } + assert _observability.build_telemetry_error_attributes(None) == {} def test_extract_error_attributes_with_error_info(): - """Proves that _extract_error_attributes extracts domain, error.type, and metadata from ErrorInfo.""" + """Proves that build_telemetry_error_attributes extracts domain, error.type, and metadata from ErrorInfo.""" import types error_info = types.SimpleNamespace( @@ -804,8 +810,9 @@ def test_extract_error_attributes_with_error_info(): }, ) exc = types.SimpleNamespace(error_info=error_info) - attrs = google.api_core.gapic_v1.method._extract_error_attributes(exc) + attrs = _observability.build_telemetry_error_attributes(exc) assert attrs == { + "rpc.response.status_code": "SimpleNamespace", "gcp.errors.domain": "googleapis.com", "error.type": "SERVICE_DISABLED", "gcp.errors.metadata.service": "secretmanager.googleapis.com", @@ -845,17 +852,18 @@ def test_wrap_method_otel_tracing_records_gcp_error_attributes(mock_otel): def test_extract_error_attributes_variations(): - """Proves that _extract_error_attributes handles __cause__, gRPC error details parsing, and direct fallbacks.""" + """Proves that build_telemetry_error_attributes handles __cause__, gRPC error details parsing, and direct fallbacks.""" import types - from google.api_core.gapic_v1.method import _extract_error_attributes + from google.api_core._observability import build_telemetry_error_attributes # 1. __cause__ attribute fallback inner_err = types.SimpleNamespace( error_info=types.SimpleNamespace(domain="d", reason="r", metadata={"k": "v"}) ) outer_err = types.SimpleNamespace(__cause__=inner_err) - assert _extract_error_attributes(outer_err) == { + assert build_telemetry_error_attributes(outer_err) == { + "rpc.response.status_code": "SimpleNamespace", "gcp.errors.domain": "d", "error.type": "r", "gcp.errors.metadata.k": "v", @@ -871,7 +879,8 @@ def test_extract_error_attributes_variations(): "google.api_core.exceptions._parse_grpc_error_details", return_value=(None, error_info), ): - assert _extract_error_attributes(exc_with_call) == { + assert build_telemetry_error_attributes(exc_with_call) == { + "rpc.response.status_code": "SimpleNamespace", "gcp.errors.domain": "parse_d", "error.type": "parse_r", "gcp.errors.metadata.foo": "bar", @@ -885,14 +894,18 @@ def test_extract_error_attributes_variations(): "google.api_core.exceptions._parse_grpc_error_details", side_effect=ValueError("bad proto"), ): - assert _extract_error_attributes(exc_with_resp) == { - "error.type": "SimpleNamespace" + assert build_telemetry_error_attributes(exc_with_resp) == { + "rpc.response.status_code": "SimpleNamespace", + "error.type": "SimpleNamespace", } # 4. error_info with empty domain, empty reason, empty metadata error_info_empty = types.SimpleNamespace(domain="", reason="", metadata=None) exc_empty = types.SimpleNamespace(error_info=error_info_empty) - assert _extract_error_attributes(exc_empty) == {"error.type": "SimpleNamespace"} + assert build_telemetry_error_attributes(exc_empty) == { + "rpc.response.status_code": "SimpleNamespace", + "error.type": "SimpleNamespace", + } # 5. else fallback where target_exc directly has domain, reason, and metadata exc_fallback = types.SimpleNamespace( @@ -900,7 +913,8 @@ def test_extract_error_attributes_variations(): reason="fallback_r", metadata={"f_key": 42}, ) - assert _extract_error_attributes(exc_fallback) == { + assert build_telemetry_error_attributes(exc_fallback) == { + "rpc.response.status_code": "SimpleNamespace", "gcp.errors.domain": "fallback_d", "error.type": "fallback_r", "gcp.errors.metadata.f_key": "42", @@ -912,34 +926,39 @@ def test_extract_error_attributes_variations(): reason="", metadata={}, ) - assert _extract_error_attributes(exc_fallback_empty) == { - "error.type": "SimpleNamespace" + assert build_telemetry_error_attributes(exc_fallback_empty) == { + "rpc.response.status_code": "SimpleNamespace", + "error.type": "SimpleNamespace", } # 7. status.message extraction from .message attribute exc_with_msg = types.SimpleNamespace(message="api call failed") - assert _extract_error_attributes(exc_with_msg) == { + assert build_telemetry_error_attributes(exc_with_msg) == { + "rpc.response.status_code": "SimpleNamespace", "error.type": "SimpleNamespace", "status.message": "api call failed", } # 8. status.message extraction from .details() callable (e.g. gRPC RpcError) exc_with_details = types.SimpleNamespace(details=lambda: "rpc deadline exceeded") - assert _extract_error_attributes(exc_with_details) == { + assert build_telemetry_error_attributes(exc_with_details) == { + "rpc.response.status_code": "SimpleNamespace", "error.type": "SimpleNamespace", "status.message": "rpc deadline exceeded", } # 9. status.message extraction from Exception string representation exc_standard = ValueError("invalid argument passed") - assert _extract_error_attributes(exc_standard) == { + assert build_telemetry_error_attributes(exc_standard) == { + "rpc.response.status_code": "ValueError", "error.type": "ValueError", "status.message": "invalid argument passed", } # 10. Exception with empty message string does not populate status.message exc_empty_msg = ValueError("") - assert _extract_error_attributes(exc_empty_msg) == { + assert build_telemetry_error_attributes(exc_empty_msg) == { + "rpc.response.status_code": "ValueError", "error.type": "ValueError", } @@ -947,13 +966,15 @@ def test_extract_error_attributes_variations(): import asyncio exc_cancelled_with_msg = asyncio.CancelledError("async task cancelled") - assert _extract_error_attributes(exc_cancelled_with_msg) == { + assert build_telemetry_error_attributes(exc_cancelled_with_msg) == { + "rpc.response.status_code": "CancelledError", "error.type": "CancelledError", "status.message": "async task cancelled", } exc_cancelled_empty = asyncio.CancelledError() - assert _extract_error_attributes(exc_cancelled_empty) == { + assert build_telemetry_error_attributes(exc_cancelled_empty) == { + "rpc.response.status_code": "CancelledError", "error.type": "CancelledError", } @@ -966,12 +987,14 @@ def test_extract_error_attributes_variations(): "POST http://localhost:7469/v1beta1/sequences/23: Persistent outage", response=http_resp_503, ) - assert _extract_error_attributes(http_exc) == { + assert build_telemetry_error_attributes(http_exc) == { + "rpc.response.status_code": "UNAVAILABLE", "error.type": "503", "status.message": "Persistent outage", } retry_http_exc = exceptions.RetryError("Deadline of 0.1s exceeded", cause=http_exc) - assert _extract_error_attributes(retry_http_exc) == { + assert build_telemetry_error_attributes(retry_http_exc) == { + "rpc.response.status_code": "UNAVAILABLE", "error.type": "503", "status.message": "Persistent outage", } @@ -989,7 +1012,8 @@ def test_extract_error_attributes_variations(): "metadata": {"service": "showcase.googleapis.com"}, }, ) - assert _extract_error_attributes(http_exc_dict_info) == { + assert build_telemetry_error_attributes(http_exc_dict_info) == { + "rpc.response.status_code": "BAD_REQUEST", "gcp.errors.domain": "googleapis.com", "error.type": "RATE_LIMIT_EXCEEDED", "gcp.errors.metadata.service": "showcase.googleapis.com", @@ -1005,7 +1029,8 @@ def test_extract_error_attributes_variations(): "gRPC transport error", response=grpc_call_like, ) - assert _extract_error_attributes(grpc_exc_with_resp) == { + assert build_telemetry_error_attributes(grpc_exc_with_resp) == { + "rpc.response.status_code": "UNAVAILABLE", "error.type": "UNAVAILABLE", "status.message": "gRPC transport error", } diff --git a/packages/google-api-core/tests/unit/test_observability.py b/packages/google-api-core/tests/unit/test_observability.py index 7cd31e95070f..f6d0ee107134 100644 --- a/packages/google-api-core/tests/unit/test_observability.py +++ b/packages/google-api-core/tests/unit/test_observability.py @@ -1633,3 +1633,175 @@ def test_suppressing_interceptors_define_slots(): assert _observability._AsyncSuppressingClientInterceptor.__slots__ == () assert "__dict__" not in _observability._SuppressingClientInterceptor.__dict__ assert "__dict__" not in _observability._AsyncSuppressingClientInterceptor.__dict__ + + +def test_has_opentelemetry_caching(monkeypatch): + """Proves has_opentelemetry caches True and False states across calls.""" + monkeypatch.setattr(_observability, "_HAS_OPENTELEMETRY", None) + monkeypatch.setitem(sys.modules, "opentelemetry", types.ModuleType("opentelemetry")) + monkeypatch.setitem( + sys.modules, + "opentelemetry.trace", + types.ModuleType("opentelemetry.trace"), + ) + + assert _observability.has_opentelemetry() is True + assert _observability._HAS_OPENTELEMETRY is True + # Cached call returns True even if sys.modules changes + monkeypatch.setitem(sys.modules, "opentelemetry.trace", None) + assert _observability.has_opentelemetry() is True + + # Reset sentinel and test when ImportError occurs + monkeypatch.setattr(_observability, "_HAS_OPENTELEMETRY", None) + assert _observability.has_opentelemetry() is False + assert _observability._HAS_OPENTELEMETRY is False + + # Cached call returns False even if opentelemetry.trace becomes available + monkeypatch.setitem( + sys.modules, + "opentelemetry.trace", + types.ModuleType("opentelemetry.trace"), + ) + assert _observability.has_opentelemetry() is False + + +@pytest.mark.parametrize( + "trace_flags,expected_sampled", + [ + pytest.param( + types.SimpleNamespace(sampled=True), True, id="flag_obj_sampled_true" + ), + pytest.param( + types.SimpleNamespace(sampled=False), + False, + id="flag_obj_sampled_false", + ), + pytest.param(1, True, id="int_bitmask_sampled_true"), + pytest.param(0, False, id="int_bitmask_sampled_false"), + pytest.param(2, False, id="int_bitmask_other_bit_unsampled"), + pytest.param(None, False, id="none_trace_flags"), + ], +) +def test_extract_trace_context_explicit_span_flags(trace_flags, expected_sampled): + """Proves extract_trace_context formats hex IDs and resolves trace_flags.""" + span = mock.Mock(spec_set=["get_span_context"]) + span.get_span_context.return_value = types.SimpleNamespace( + is_valid=True, + trace_id=0x0123456789ABCDEF0123456789ABCDEF, + span_id=0x0123456789ABCDEF, + trace_flags=trace_flags, + ) + + expected = { + "trace_id": "0123456789abcdef0123456789abcdef", + "span_id": "0123456789abcdef", + "logging.googleapis.com/trace": "0123456789abcdef0123456789abcdef", + "logging.googleapis.com/spanId": "0123456789abcdef", + "logging.googleapis.com/trace_sampled": expected_sampled, + } + assert _observability.extract_trace_context(span) == expected + + +def test_extract_trace_context_invalid_or_broken_span(): + """Proves extract_trace_context returns {} on broken spans or non-int IDs.""" + bad_span = mock.Mock(spec_set=["get_span_context"]) + bad_span.get_span_context.side_effect = RuntimeError("broken span") + assert _observability.extract_trace_context(bad_span) == {} + + bad_ctx_span = mock.Mock(spec_set=["get_span_context"]) + bad_ctx_span.get_span_context.return_value = types.SimpleNamespace( + is_valid=True, trace_id="not_an_int", span_id="not_an_int" + ) + assert _observability.extract_trace_context(bad_ctx_span) == {} + + +def test_extract_trace_context_ambient_fallback(monkeypatch, mock_otel): + """Proves extract_trace_context falls back to get_current_span().""" + monkeypatch.setattr(_observability, "_HAS_OPENTELEMETRY", True) + mock_otel.span.get_span_context.return_value = types.SimpleNamespace( + is_valid=True, + trace_id=0x0123456789ABCDEF0123456789ABCDEF, + span_id=0x0123456789ABCDEF, + trace_flags=types.SimpleNamespace(sampled=True), + ) + + res = _observability.extract_trace_context(None) + assert res["trace_id"] == "0123456789abcdef0123456789abcdef" + assert res["logging.googleapis.com/trace_sampled"] is True + + # Ambient get_current_span raising an exception returns {} + mock_otel.trace.get_current_span.side_effect = RuntimeError("ambient error") + assert _observability.extract_trace_context(None) == {} + + # OTel not installed returns {} + monkeypatch.setattr(_observability, "_HAS_OPENTELEMETRY", False) + assert _observability.extract_trace_context(None) == {} + + +@pytest.mark.parametrize( + "exc,kwargs,expected", + [ + pytest.param(None, {}, {}, id="none_exc_no_kwargs"), + pytest.param( + None, + {"status_code": "503"}, + { + "rpc.response.status_code": "503", + "error.type": "503", + "status.message": "HTTP 503", + }, + id="none_exc_with_status_code", + ), + pytest.param( + None, + {"status_code": "503", "status_message": "Service Unavailable"}, + { + "rpc.response.status_code": "503", + "error.type": "503", + "status.message": "Service Unavailable", + }, + id="none_exc_with_status_code_and_message", + ), + pytest.param( + { + "domain": "googleapis.com", + "reason": "RATE_LIMIT_EXCEEDED", + "metadata": {"quota": "100"}, + }, + {"status_message": "Quota exceeded"}, + { + "rpc.response.status_code": "dict", + "gcp.errors.domain": "googleapis.com", + "error.type": "RATE_LIMIT_EXCEEDED", + "gcp.errors.metadata.quota": "100", + "status.message": "Quota exceeded", + }, + id="dict_error_info", + ), + ], +) +def test_build_telemetry_error_attributes_variants(exc, kwargs, expected): + """Proves build_telemetry_error_attributes handles None and dict inputs.""" + assert _observability.build_telemetry_error_attributes(exc, **kwargs) == expected + + +def test_build_telemetry_error_attributes_include_exception(): + """Proves include_exception=True formats real exception stacktraces.""" + try: + raise RuntimeError("") + except RuntimeError as caught_exc: + exc_empty = caught_exc + + attrs_inc = _observability.build_telemetry_error_attributes( + exc_empty, + status_code="UNAVAILABLE", + status_message="backend down", + include_exception=True, + ) + assert attrs_inc["rpc.response.status_code"] == "UNAVAILABLE" + assert attrs_inc["error.type"] == "UNAVAILABLE" + assert attrs_inc["status.message"] == "backend down" + assert attrs_inc["exception.type"] == "RuntimeError" + assert attrs_inc["exception.message"] == "" + assert "Traceback (most recent call last):" in attrs_inc["exception.stacktrace"] + assert "raise RuntimeError" in attrs_inc["exception.stacktrace"]