feat(google-api-core): extend ErrorInfo dict accessors, centralize status code extraction, and update StructuredLogFormatter

packages/google-api-core/google/api_core/client_logging.py

@@ -1,6 +1,7 @@
 import json
 import logging
 import os
+from collections.abc import Mapping
 from typing import List, Optional
 
 _LOGGING_INITIALIZED = False
@@ -10,6 +11,7 @@
 #
 # TODO(https://github.com/googleapis/python-api-core/issues/761): Update this list to support additional logging fields.
 _recognized_logging_fields = [
+    # Application payload debugging fields:
     "httpRequest",
     "rpcName",
     "serviceName",
@@ -21,8 +23,31 @@
     "metadata",
     "retryAttempt",
     "httpResponse",
+    # Diagnostic error and trace correlation fields:
+    "url.full",
+    "rpc.method",
+    "exception.type",
+    "exception.message",
+    "exception.stacktrace",
+    "rpc.system.name",
+    "rpc.response.status_code",
+    "error.type",
+    "status.message",
+    "gcp.errors.domain",
+    "gcp.errors.metadata",
+    "trace_id",
+    "span_id",
+    "logging.googleapis.com/trace",
+    "logging.googleapis.com/spanId",
+    "logging.googleapis.com/trace_sampled",
+    # Dynamic metadata prefixes:
+    "gcp.errors.metadata.*",
 ]  # Additional fields to be Logged.
 
+# Pre-compiled allowlists for O(1) exact lookups and C-level prefix matching.
+_EXACT_FIELDS = frozenset(f for f in _recognized_logging_fields if not f.endswith("*"))
+_PREFIX_FIELDS = tuple(f[:-1] for f in _recognized_logging_fields if f.endswith("*"))
+
 
 def logger_configured(logger) -> bool:
     """Determines whether `logger` has non-default configuration
@@ -136,8 +161,22 @@ def format(self, record: logging.LogRecord):
             "message": record.getMessage(),
         }
 
-        for field_name in _recognized_logging_fields:
-            value = getattr(record, field_name, None)
+        # Extract exact allowed fields via getattr() (supports @property and class-level attributes)
+        for key in _EXACT_FIELDS:
+            value = getattr(record, key, None)
             if value is not None:
-                log_obj[field_name] = value
+                log_obj[key] = value
+
+        # Extract dynamic wildcard prefix fields (e.g. gcp.errors.metadata.*) from instance __dict__
+        for key, value in vars(record).items():
+            if value is not None and key.startswith(_PREFIX_FIELDS):
+                log_obj[key] = value
+
+        # Unpack record.json_fields last so explicit user-provided structured fields
+        # take precedence over default LogRecord attributes (consistent with google-cloud-logging).
+        json_fields = getattr(record, "json_fields", None)
+        if isinstance(json_fields, Mapping):
+            for key, value in json_fields.items():
+                if value is not None:
+                    log_obj[key] = value
         return json.dumps(log_obj)

packages/google-api-core/google/api_core/exceptions.py

@@ -20,6 +20,7 @@
 
 from __future__ import absolute_import, unicode_literals
 
+import http
 import http.client
 import warnings
 from typing import Dict, Optional
@@ -177,7 +178,11 @@ def reason(self):
         Returns:
             Union[str, None]: An optional string containing reason of the error.
         """
-        return self._error_info.reason if self._error_info else None
+        if not self._error_info:
+            return None
+        if isinstance(self._error_info, dict):
+            return self._error_info.get("reason")
+        return self._error_info.reason
 
     @property
     def domain(self):
@@ -189,7 +194,11 @@ def domain(self):
         Returns:
             Union[str, None]: An optional string containing a logical grouping to which the "reason" belongs.
         """
-        return self._error_info.domain if self._error_info else None
+        if not self._error_info:
+            return None
+        if isinstance(self._error_info, dict):
+            return self._error_info.get("domain")
+        return self._error_info.domain
 
     @property
     def metadata(self):
@@ -201,7 +210,11 @@ def metadata(self):
         Returns:
             Union[Dict[str, str], None]: An optional object containing structured details about the error.
         """
-        return self._error_info.metadata if self._error_info else None
+        if not self._error_info:
+            return None
+        if isinstance(self._error_info, dict):
+            return self._error_info.get("metadata")
+        return self._error_info.metadata
 
     @property
     def errors(self):
@@ -637,6 +650,62 @@ def from_grpc_status(status_code, message, **kwargs):
     return error
 
 
+def _extract_status_code(exc: BaseException | None, default: str | None = None) -> str:
+    """Extract canonical status code name string from an exception.
+
+    Status code name strings are resolved by inspecting the following:
+    * Chained exceptions: Unwraps RetryError or __cause__ to the root error.
+    * Enum & code attributes: Inspects .grpc_status_code on GoogleAPICallError
+      or .code on gRPC/HTTP errors.
+    * Integer status codes: Maps raw gRPC integer status codes or HTTP status
+      codes to canonical enum names.
+    * Fallback: Uses ``default`` if provided, otherwise defaults to the
+      exception class name.
+
+    Args:
+        exc (BaseException | None): The exception to extract the status code
+            name from.
+        default (str | None): Optional fallback status code string when
+            ``exc`` carries no status code.
+
+    Returns:
+        str: The canonical status code name (e.g. "NOT_FOUND", "UNAVAILABLE",
+            "BAD_REQUEST") or class name.
+    """
+    if exc is None:
+        return default or ""
+
+    # 1. Unwrap chained exceptions: unwrap RetryError or __cause__ to the root failure
+    target = getattr(exc, "cause", None) or getattr(exc, "__cause__", None) or exc
+
+    # 2. Check enum & code attributes: .grpc_status_code enum or callable/non-callable .code
+    status = getattr(target, "grpc_status_code", None)
+    if status is None and hasattr(target, "code"):
+        try:
+            status = target.code() if callable(target.code) else target.code
+        # Defensive fallback: observability status extraction must never raise
+        # if a third-party or mock exception implements a failing .code() method.
+        except Exception:
+            status = None
+
+    name = getattr(status, "name", None)
+    if name:
+        return str(name)
+
+    # 3. Check integer status codes: map raw gRPC integer codes or HTTP status codes to canonical names
+    if isinstance(status, int):
+        if status in _INT_TO_GRPC_CODE:
+            grpc_code = _INT_TO_GRPC_CODE[status]
+            return getattr(grpc_code, "name", str(grpc_code))
+        try:
+            return http.HTTPStatus(status).name
+        except ValueError:
+            return str(status)
+
+    # 4. Fallback: default to explicit default or exception class name for standard Python errors
+    return default or target.__class__.__name__
+
+
 def _is_informative_grpc_error(rpc_exc):
     return hasattr(rpc_exc, "code") and hasattr(rpc_exc, "details")
 

packages/google-api-core/google/api_core/gapic_v1/method.py

@@ -23,7 +23,7 @@
 import functools
 from typing import Any, List, Optional, Tuple
 
-from google.api_core import _observability, grpc_helpers
+from google.api_core import _observability, exceptions, 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,50 +126,6 @@ def _extract_rpc_identity(
     return method_str, service, method
 
 
-def _extract_status_code(exc: Optional[BaseException]) -> str:
-    """Extract canonical status code name string from an exception.
-
-    Status code name strings are resolved by inspecting the following locations:
-    * Chained exceptions: Unwraps RetryError or __cause__ to the root exception.
-    * Enum & code attributes: Inspects .grpc_status_code on GoogleAPICallError or .code on gRPC errors.
-    * Integer status codes: Maps raw gRPC integer status codes to canonical enum names.
-    * Fallback: Defaults to the exception class name for standard Python errors.
-
-    Args:
-        exc (Optional[BaseException]): The exception to extract the status code name from.
-
-    Returns:
-        str: The canonical status code name (e.g. "NOT_FOUND", "UNAVAILABLE") or class name.
-    """
-    if exc is None:
-        return ""
-
-    # 1. Unwrap chained exceptions: unwrap RetryError or __cause__ to the root failure
-    target = getattr(exc, "cause", None) or getattr(exc, "__cause__", None) or exc
-
-    # 2. Check enum & code attributes: .grpc_status_code enum or callable/non-callable .code
-    status = getattr(target, "grpc_status_code", None)
-    if status is None and hasattr(target, "code"):
-        try:
-            status = target.code() if callable(target.code) else target.code
-        except Exception:
-            status = None
-
-    name = getattr(status, "name", None)
-    if name:
-        return str(name)
-
-    # 3. Check integer status codes: map raw gRPC integer status codes to canonical enum names
-    if isinstance(status, int):
-        from google.api_core import exceptions
-
-        status = exceptions._INT_TO_GRPC_CODE.get(status, status)
-        return getattr(status, "name", str(status))
-
-    # 4. Fallback: default to the exception class name for standard Python errors
-    return target.__class__.__name__
-
-
 def _extract_error_attributes(exc: Optional[BaseException]) -> dict[str, Any]:
     """Extract gcp.errors.* and error.type attributes from an exception.
 
@@ -206,8 +162,6 @@ def _extract_error_attributes(exc: Optional[BaseException]) -> dict[str, Any]:
         )
         if rpc_call is not None and hasattr(rpc_call, "trailing_metadata"):
             try:
-                from google.api_core import exceptions
-
                 _, error_info = exceptions._parse_grpc_error_details(rpc_call)
             except Exception:
                 pass
@@ -239,7 +193,7 @@ def _extract_error_attributes(exc: Optional[BaseException]) -> dict[str, Any]:
     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 = _extract_status_code(target_exc)
+        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():
@@ -428,7 +382,7 @@ def _trace_span(self):
             except BaseException as exc:
                 if span is not None and hasattr(span, "set_attribute"):
                     span.set_attribute(
-                        "rpc.response.status_code", _extract_status_code(exc)
+                        "rpc.response.status_code", exceptions._extract_status_code(exc)
                     )
                     for k, v in _extract_error_attributes(exc).items():
                         span.set_attribute(k, v)

packages/google-api-core/tests/unit/gapic/test_method.py

@@ -844,56 +844,6 @@ def test_wrap_method_otel_tracing_records_gcp_error_attributes(mock_otel):
     mock_otel.span.set_attribute.assert_any_call("status.message", "quota exceeded")
 
 
-def test_extract_status_code_variations():
-    """Proves that _extract_status_code handles grpc status, callable/non-callable codes, ints, and exceptions."""
-    import types
-
-    from google.api_core.gapic_v1.method import _extract_status_code
-
-    # 1. grpc_status exists but has no name or name is None
-    exc1 = types.SimpleNamespace(grpc_status_code=types.SimpleNamespace(name=None))
-    assert _extract_status_code(exc1) == "SimpleNamespace"
-
-    # 2. callable code_fn returns object with name
-    exc2 = types.SimpleNamespace(code=lambda: types.SimpleNamespace(name="CANCELLED"))
-    assert _extract_status_code(exc2) == "CANCELLED"
-
-    # 3. callable code_fn returns object without name
-    exc3 = types.SimpleNamespace(code=lambda: types.SimpleNamespace(name=None))
-    assert _extract_status_code(exc3) == "SimpleNamespace"
-
-    # 4. callable code_fn raises Exception
-    def raising_code():
-        raise RuntimeError("boom")
-
-    exc4 = types.SimpleNamespace(code=raising_code)
-    assert _extract_status_code(exc4) == "SimpleNamespace"
-
-    # 5. non-callable code_fn with name
-    exc5 = types.SimpleNamespace(code=types.SimpleNamespace(name="DEADLINE_EXCEEDED"))
-    assert _extract_status_code(exc5) == "DEADLINE_EXCEEDED"
-
-    # 6. non-callable code_fn that is an int in _INT_TO_GRPC_CODE (5 -> NOT_FOUND)
-    exc6 = types.SimpleNamespace(code=5)
-    assert _extract_status_code(exc6) == "NOT_FOUND"
-
-    # 7. non-callable code_fn that is an int not in _INT_TO_GRPC_CODE (999)
-    exc7 = types.SimpleNamespace(code=999)
-    assert _extract_status_code(exc7) == "999"
-
-    # 8. non-callable code_fn that is not an int and has no name
-    exc8 = types.SimpleNamespace(code="unknown_code")
-    assert _extract_status_code(exc8) == "SimpleNamespace"
-
-    # 9. None exception
-    assert _extract_status_code(None) == ""
-
-    # 10. __cause__ chaining fallback
-    inner_exc = types.SimpleNamespace(code=5)
-    outer_exc = types.SimpleNamespace(__cause__=inner_exc)
-    assert _extract_status_code(outer_exc) == "NOT_FOUND"
-
-
 def test_extract_error_attributes_variations():
     """Proves that _extract_error_attributes handles __cause__, gRPC error details parsing, and direct fallbacks."""
     import types