[chalmerlowe]

Problem / Context

To support uniform structured error logging and OpenTelemetry telemetry across both gRPC and REST transports in google-api-core, several foundational utilities need to handle both transport formats consistently:

1. REST ErrorInfo Accessors: GoogleAPICallError.reason, .domain, and .metadata previously assumed self._error_info was always a protobuf ErrorInfo message (as returned by gRPC google.rpc.status_pb2.Status). For REST JSON error payloads, ErrorInfo entries in error["details"] are parsed as standard Python dict objects, causing attribute access (self._error_info.reason) to fail or return None.
2. Duplicate Status Code Extraction: Status code normalization across GoogleAPICallError and raw grpc.RpcError instances (_extract_status_code) lived inside google.api_core.gapic_v1.method, preventing reuse by transport-level observability helpers.
3. Structured Log Field Allowlist: StructuredLogFormatter only recognized legacy camelCase fields and matched fields via exact set membership (in _recognized_logging_fields). It did not unpack record.json_fields or support OpenTelemetry semantic conventions (such as rpc.*, http.*, url.*, server.*, error.type, exception.*, and dynamic gcp.errors.metadata.* keys).

Solution

• google/api_core/exceptions.py:
  • Updated GoogleAPICallError.reason, .domain, and .metadata to inspect isinstance(self._error_info, dict) first before falling back to protobuf attribute access.
  • Moved _extract_status_code(exc, default=None) to google.api_core.exceptions so both gapic_v1.method and transport observability helpers can share a single status code normalization implementation.
• google/api_core/gapic_v1/method.py:
  • Aliased _extract_status_code = exceptions._extract_status_code to preserve backward compatibility while eliminating duplicate code.
• google/api_core/client_logging.py:
  • Extended _recognized_logging_fields with OpenTelemetry semantic convention attributes, Google Cloud error domain/metadata fields (gcp.errors.domain, gcp.errors.metadata, gcp.errors.metadata.*), and trace correlation keys (logging.googleapis.com/trace, logging.googleapis.com/spanId, logging.googleapis.com/trace_sampled).
  • Pre-compiled _EXACT_FIELDS (frozenset) and _PREFIX_FIELDS (tuple) at module import time and updated StructuredLogFormatter.format to unpack record.json_fields and filter recognized fields in a single pass.
• Unit Tests:
  • Added unit tests in tests/unit/test_exceptions.py and tests/unit/test_client_logging.py achieving 100% line and branch coverage on modified modules.

[gemini-code-assist]

Code Review

This pull request enhances client logging and exception handling in google-api-core. It expands recognized logging fields, optimizes field extraction using pre-compiled allowlists, and updates StructuredLogFormatter to unpack record.json_fields and extract allowed fields from vars(record). Additionally, it updates exception properties to support dictionary-based error info and centralizes the _extract_status_code utility. The review feedback suggests reversing the processing order in StructuredLogFormatter so that custom fields in json_fields take precedence over automatically extracted attributes from vars(record).

[gemini-code-assist]

Currently, record.json_fields is unpacked first, and then the allowed fields from vars(record) are extracted and added to log_obj. If a key exists in both json_fields and vars(record) (for example, if a user explicitly overrides a standard/allowed field in json_fields with a custom value), the value from vars(record) will overwrite the custom value from json_fields because step 2 runs after step 1.

To ensure that explicitly provided custom fields in json_fields take precedence over automatically extracted attributes from vars(record), we should reverse the order of these two steps.

Suggested change

	# 1. Unpack record.json_fields (keeping api_core and google-cloud-logging formatters consistent)
	json_fields = getattr(record, "json_fields", None)
	if isinstance(json_fields, collections.abc.Mapping):
	for key, value in json_fields.items():
	if value is not None:
	log_obj[key] = value
	# 2. Extract allowed fields in a single pass over vars(record)
	for key, value in vars(record).items():
	if value is not None and (
	key in _EXACT_FIELDS or key.startswith(_PREFIX_FIELDS)
	):
	log_obj[key] = value
	# 1. Extract allowed fields in a single pass over vars(record)
	for key, value in vars(record).items():
	if value is not None and (
	key in _EXACT_FIELDS or key.startswith(_PREFIX_FIELDS)
	):
	log_obj[key] = value
	# 2. Unpack record.json_fields (keeping api_core and google-cloud-logging formatters consistent)
	json_fields = getattr(record, "json_fields", None)
	if isinstance(json_fields, collections.abc.Mapping):
	for key, value in json_fields.items():
	if value is not None:
	log_obj[key] = value