From ca5d7114c0c82fdf183711b7d0b931548c8489e4 Mon Sep 17 00:00:00 2001 From: chalmer lowe Date: Fri, 6 Nov 2026 11:37:55 -0500 Subject: [PATCH] feat(api_core): add method-level terminal error logging in _GapicCallable --- .../google/api_core/gapic_v1/method.py | 128 ++++++++++++++++-- .../tests/asyncio/gapic/test_method_async.py | 53 ++++++++ .../tests/unit/gapic/test_method.py | 107 +++++++++++++++ 3 files changed, 278 insertions(+), 10 deletions(-) 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 e119ab414a00..f3f1973d8130 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 @@ -20,16 +20,27 @@ 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() @@ -255,9 +266,110 @@ 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: + extra_payload: dict[str, Any] = { + "rpcName": self._method_name, + "rpc.system.name": self._rpc_system, + **error_attrs, + **_observability.extract_trace_context(span), + } + if is_cancelled: + extra_payload.pop("exception.stacktrace", None) + # 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 in an isolated try/except block so a + # failing custom Span implementation cannot skip terminal error logging below. + 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() @@ -273,11 +385,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..90ea4a002038 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 @@ -14,6 +14,7 @@ import asyncio import datetime +import logging try: from unittest import mock @@ -630,3 +631,55 @@ 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,expected_exc_info,expected_status,expect_stacktrace", + [ + pytest.param( + lambda: exceptions.NotFound("Secret not found"), + logging.ERROR, + True, + "NOT_FOUND", + True, + id="terminal_error", + ), + pytest.param( + lambda: asyncio.CancelledError("Task cancelled"), + logging.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, + expected_exc_info, + expected_status, + expect_stacktrace, +): + """Proves that _AsyncGapicCallable emits ERROR on terminal failure and DEBUG without traceback on asyncio.CancelledError.""" + 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] == expected_level + 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 6c2703862fc3..14a390db8643 100644 --- a/packages/google-api-core/tests/unit/gapic/test_method.py +++ b/packages/google-api-core/tests/unit/gapic/test_method.py @@ -12,7 +12,9 @@ # See the License for the specific language governing permissions and # limitations under the License. +import concurrent.futures import datetime +import logging import sys import types from unittest import mock @@ -848,3 +850,108 @@ 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.""" + 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.""" + 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()