feat(api_core): add method-level terminal error logging in _GapicCallable

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__(

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

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()