Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# -*- coding: utf-8 -*-
2# Copyright 2026 Google LLC
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15#
17"""OpenTelemetry helpers for resolving and instantiating interceptors."""
19from __future__ import annotations
21import contextlib
22import urllib.parse
23from typing import TYPE_CHECKING, Any, Callable, Sequence
25from google.api_core import _feature_gating_helpers
26from google.api_core.client_options import ClientOptions
28try:
29 from opentelemetry.instrumentation.utils import (
30 suppress_instrumentation as _suppress_instrumentation,
31 )
32except ImportError:
33 _suppress_instrumentation = contextlib.nullcontext
35if TYPE_CHECKING:
36 # flake8: grpc, trace, and ClientInterceptor are imported only for static analysis and type annotations
37 # The 'noqa: F401' comment avoids flake8 "imported but not used" errors.
38 import grpc # noqa: F401
39 import opentelemetry.trace # noqa: F401
41 from google.api_core.grpc_helpers import ClientInterceptor # noqa: F401
43_TRACER_PROVIDER = "tracer_provider"
46def is_otel_capabilities_enabled(
47 client_options: ClientOptions | dict[str, Any] | None = None,
48 env_var: str = "GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED",
49) -> bool:
50 """Checks if OTel capabilities are enabled and installed.
52 Args:
53 client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options
54 object or dictionary.
55 env_var (str): The environment variable to check for enablement.
57 Returns:
58 bool: True if enabled and installed, False otherwise.
59 """
60 is_tracing_enabled = _feature_gating_helpers.resolve_feature_flags(
61 env_var=env_var,
62 feature_key=_TRACER_PROVIDER,
63 configuration=client_options,
64 )
66 if is_tracing_enabled:
67 try:
68 import opentelemetry.instrumentation.grpc as otel_grpc # type: ignore[import-not-found] # noqa: F401
70 return True
71 except ImportError:
72 pass
74 return False
77def _extract_endpoint_attributes(
78 client_options: ClientOptions | dict[str, Any] | None = None,
79) -> dict[str, Any]:
80 """Extracts server.address, server.port (if non-default), and url.domain from client options if present.
82 Args:
83 client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options
84 object or dictionary.
86 Returns:
87 dict[str, Any]: A dictionary containing url.domain and, if an api_endpoint is configured,
88 server.address and non-default server.port.
89 """
90 attrs: dict[str, Any] = {}
91 endpoint = None
92 universe_domain = None
94 if isinstance(client_options, dict):
95 endpoint = client_options.get("api_endpoint")
96 universe_domain = client_options.get("universe_domain")
97 elif client_options is not None:
98 endpoint = getattr(client_options, "api_endpoint", None)
99 universe_domain = getattr(client_options, "universe_domain", None)
101 attrs["url.domain"] = universe_domain or "googleapis.com"
103 if endpoint and isinstance(endpoint, str):
104 target = endpoint if "//" in endpoint else f"//{endpoint}"
105 parsed = None
106 hostname = None
107 port = None
108 try:
109 parsed = urllib.parse.urlsplit(target)
110 hostname = parsed.hostname
111 port = parsed.port
112 except ValueError:
113 pass
115 if hostname:
116 attrs["server.address"] = hostname
117 if port and parsed:
118 scheme = parsed.scheme.lower()
119 is_default_port = (port == 443 and scheme in ("https", "")) or (
120 port == 80 and scheme == "http"
121 )
122 if not is_default_port:
123 attrs["server.port"] = port
124 return attrs
127def _make_grpc_client_request_hook(
128 endpoint_attrs: dict[str, Any] | None = None,
129) -> Callable[[Any, Any], None]:
130 """Creates an OpenTelemetry gRPC client request hook with optional endpoint attributes.
132 Args:
133 endpoint_attrs (Optional[dict[str, Any]]): Optional static endpoint attributes to attach
134 to every span.
136 Returns:
137 Callable[[Any, Any], None]: The request hook callback.
138 """
139 static_attrs = dict(endpoint_attrs) if endpoint_attrs else {}
141 def client_request_hook(span: Any, request: Any) -> None:
142 if span is None or not getattr(span, "is_recording", lambda: True)():
143 return
145 # Upstream opentelemetry-instrumentation-grpc may format span names with a
146 # leading slash (e.g. "/package.Service/Method"). Normalize the span name
147 # and ensure rpc.method is always captured as the clean, fully-qualified name.
148 span_name = getattr(span, "name", None)
149 if isinstance(span_name, str) and span_name:
150 clean_method_name = span_name.lstrip("/")
151 if span_name.startswith("/") and hasattr(span, "update_name"):
152 span.update_name(clean_method_name)
153 span.set_attribute("rpc.method", clean_method_name)
155 attrs: dict[str, Any] = {
156 "rpc.system.name": "grpc",
157 }
158 if static_attrs:
159 attrs.update(static_attrs)
160 for key, value in attrs.items():
161 span.set_attribute(key, value)
163 return client_request_hook
166_grpc_client_request_hook = _make_grpc_client_request_hook()
169def _grpc_client_response_hook(span: Any, response: Any) -> None:
170 """OpenTelemetry gRPC client response hook to record successful response status.
172 Upstream ``opentelemetry-instrumentation-grpc`` sets the integer status code
173 ``rpc.grpc.status_code`` (e.g. 0), but does not record the modern string status
174 ``rpc.response.status_code`` (e.g. "OK") required by Cloud Trace and current
175 OpenTelemetry semantic conventions (v1.27.0+).
177 This hook enriches successful RPC attempt spans with ``rpc.response.status_code = "OK"``.
178 Errors and non-OK statuses are handled at the Tier 3 method span layer or upstream.
180 Upstream handles synchronous and asynchronous invocations differently:
181 - **Synchronous gRPC**: Upstream only invokes the response hook when an RPC call
182 succeeds. On failure, the hook is bypassed entirely.
183 - **Asynchronous gRPC**: Upstream invokes the response hook unconditionally for
184 both successes and failures (passing exception details on error). However, it
185 always marks ``span.status`` with an error status before calling the hook.
187 Because of this disparity, this hook checks ``span.status`` to guard against
188 async failure callbacks while allowing synchronous and successful asynchronous
189 calls to be marked "OK".
191 Note:
192 If upstream ``opentelemetry-instrumentation-grpc`` adds native support for
193 modern ``rpc.response.status_code`` in future releases, this hook can be retired.
195 Args:
196 span (Optional[Any]): The OpenTelemetry span.
197 response (Any): The gRPC response object or details.
198 """
199 if span is None or not getattr(span, "is_recording", lambda: False)():
200 return
202 # Guard against upstream async calls that invoke this hook on failures.
203 status = getattr(span, "status", None)
204 status_code = getattr(status, "status_code", None)
205 if (
206 getattr(status_code, "name", None) == "ERROR"
207 or getattr(status_code, "value", None) == 2
208 ):
209 return
211 span.set_attribute("rpc.response.status_code", "OK")
214def _get_tracer_provider(
215 client_options: ClientOptions | dict[str, Any] | None = None,
216) -> opentelemetry.trace.TracerProvider | None:
217 """Extracts the OpenTelemetry tracer provider from client options if present.
219 Args:
220 client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options
221 object or dictionary.
223 Returns:
224 opentelemetry.trace.TracerProvider | None: The tracer provider if present,
225 None otherwise.
226 """
227 if isinstance(client_options, dict):
228 return client_options.get(_TRACER_PROVIDER)
229 elif client_options is not None:
230 return getattr(client_options, _TRACER_PROVIDER, None)
231 return None
234try:
235 # flake8: `grpc` imported conditionally for runtime interceptor base classes
236 import grpc # noqa: F811
237 from grpc import (
238 StreamStreamClientInterceptor as _SyncStreamStreamClientInterceptor,
239 )
240 from grpc import (
241 StreamUnaryClientInterceptor as _SyncStreamUnaryClientInterceptor,
242 )
243 from grpc import (
244 UnaryStreamClientInterceptor as _SyncUnaryStreamClientInterceptor,
245 )
246 from grpc import (
247 UnaryUnaryClientInterceptor as _SyncUnaryUnaryClientInterceptor,
248 )
249 from grpc.aio import (
250 StreamStreamClientInterceptor as _AsyncStreamStreamClientInterceptor,
251 )
252 from grpc.aio import (
253 StreamUnaryClientInterceptor as _AsyncStreamUnaryClientInterceptor,
254 )
255 from grpc.aio import (
256 UnaryStreamClientInterceptor as _AsyncUnaryStreamClientInterceptor,
257 )
258 from grpc.aio import (
259 UnaryUnaryClientInterceptor as _AsyncUnaryUnaryClientInterceptor,
260 )
261except ImportError: # pragma: NO COVER
262 # mypy: Fallbacks: grpc sentinel when optional grpc is not installed in the environment.
263 # and distinct empty classes avoid duplicate base class 'object' TypeError at runtime.
264 grpc = None # type: ignore[assignment]
266 class _SyncUnaryUnaryClientInterceptor: # type: ignore[no-redef]
267 __slots__ = ()
269 class _SyncUnaryStreamClientInterceptor: # type: ignore[no-redef]
270 __slots__ = ()
272 class _SyncStreamUnaryClientInterceptor: # type: ignore[no-redef]
273 __slots__ = ()
275 class _SyncStreamStreamClientInterceptor: # type: ignore[no-redef]
276 __slots__ = ()
278 class _AsyncUnaryUnaryClientInterceptor: # type: ignore[no-redef]
279 __slots__ = ()
281 class _AsyncUnaryStreamClientInterceptor: # type: ignore[no-redef]
282 __slots__ = ()
284 class _AsyncStreamUnaryClientInterceptor: # type: ignore[no-redef]
285 __slots__ = ()
287 class _AsyncStreamStreamClientInterceptor: # type: ignore[no-redef]
288 __slots__ = ()
291class _SuppressingClientInterceptor(
292 _SyncUnaryUnaryClientInterceptor,
293 _SyncUnaryStreamClientInterceptor,
294 _SyncStreamUnaryClientInterceptor,
295 _SyncStreamStreamClientInterceptor,
296):
297 """Client interceptor that suppresses redundant downstream generic auto-instrumentation spans.
299 When users enable global OpenTelemetry gRPC auto-instrumentation (e.g. GrpcInstrumentorClient),
300 the underlying channel is wrapped with generic interceptors that emit bare-bones spans.
301 This interceptor wraps downstream calls in OpenTelemetry's native `suppress_instrumentation`
302 context manager so that our feature-rich Google Cloud SDK T4 span is emitted while
303 redundant generic downstream spans are bypassed.
304 """
306 __slots__ = ()
308 def intercept_unary_unary(
309 self, continuation: Any, client_call_details: Any, request: Any
310 ) -> Any:
311 with _suppress_instrumentation():
312 return continuation(client_call_details, request)
314 def intercept_unary_stream(
315 self, continuation: Any, client_call_details: Any, request: Any
316 ) -> Any:
317 with _suppress_instrumentation():
318 return continuation(client_call_details, request)
320 def intercept_stream_unary(
321 self, continuation: Any, client_call_details: Any, request_iterator: Any
322 ) -> Any:
323 with _suppress_instrumentation():
324 return continuation(client_call_details, request_iterator)
326 def intercept_stream_stream(
327 self, continuation: Any, client_call_details: Any, request_iterator: Any
328 ) -> Any:
329 with _suppress_instrumentation():
330 return continuation(client_call_details, request_iterator)
333class _AsyncSuppressingClientInterceptor(
334 _AsyncUnaryUnaryClientInterceptor,
335 _AsyncUnaryStreamClientInterceptor,
336 _AsyncStreamUnaryClientInterceptor,
337 _AsyncStreamStreamClientInterceptor,
338):
339 """AsyncIO client interceptor that suppresses redundant downstream generic auto-instrumentation spans."""
341 __slots__ = ()
343 async def intercept_unary_unary(
344 self, continuation: Any, client_call_details: Any, request: Any
345 ) -> Any:
346 with _suppress_instrumentation():
347 return await continuation(client_call_details, request)
349 async def intercept_unary_stream(
350 self, continuation: Any, client_call_details: Any, request: Any
351 ) -> Any:
352 with _suppress_instrumentation():
353 return await continuation(client_call_details, request)
355 async def intercept_stream_unary(
356 self, continuation: Any, client_call_details: Any, request_iterator: Any
357 ) -> Any:
358 with _suppress_instrumentation():
359 return await continuation(client_call_details, request_iterator)
361 async def intercept_stream_stream(
362 self, continuation: Any, client_call_details: Any, request_iterator: Any
363 ) -> Any:
364 with _suppress_instrumentation():
365 return await continuation(client_call_details, request_iterator)
368def get_otel_interceptor(
369 client_options: ClientOptions | dict[str, Any] | None = None,
370) -> Callable[[grpc.Channel], grpc.Channel] | None:
371 """Returns an interceptor callable that wraps a sync gRPC channel with OpenTelemetry tracing.
373 Args:
374 client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options
375 object or dictionary used for feature gating and extracting the tracer provider.
377 Returns:
378 Callable[[grpc.Channel], grpc.Channel] | None: An interceptor callable if OpenTelemetry
379 tracing is enabled and installed, None otherwise.
380 """
381 if not is_otel_capabilities_enabled(client_options):
382 return None
384 import opentelemetry.instrumentation.grpc as otel_grpc # type: ignore[import-not-found]
386 endpoint_attrs = _extract_endpoint_attributes(client_options)
387 request_hook = _make_grpc_client_request_hook(endpoint_attrs)
389 interceptor: ClientInterceptor = otel_grpc.client_interceptor(
390 tracer_provider=_get_tracer_provider(client_options),
391 request_hook=request_hook,
392 response_hook=_grpc_client_response_hook,
393 )
394 suppressor = _SuppressingClientInterceptor()
396 def otel_interceptor(channel: grpc.Channel) -> grpc.Channel:
397 if grpc is not None:
398 try:
399 channel = grpc.intercept_channel(channel, suppressor)
400 except (AttributeError, TypeError):
401 # Fallback if grpc lacks intercept_channel or custom channel type rejects it.
402 pass
403 return otel_grpc.intercept_channel(channel, interceptor)
405 otel_interceptor._is_otel_interceptor = True # type: ignore[attr-defined]
406 return otel_interceptor
409def get_otel_async_interceptor(
410 client_options: ClientOptions | dict[str, Any] | None = None,
411) -> Sequence[grpc.aio.ClientInterceptor] | None:
412 """Returns async gRPC client interceptors for OpenTelemetry tracing.
414 Args:
415 client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options
416 object or dictionary used for feature gating and extracting the tracer provider.
418 Returns:
419 Sequence[grpc.aio.ClientInterceptor] | None: Instantiated OpenTelemetry async
420 client interceptors if tracing is enabled and installed, None otherwise.
421 """
422 if not is_otel_capabilities_enabled(client_options):
423 return None
425 # Ignored by mypy: Optional dependency only loaded if early-return is skipped
426 import opentelemetry.instrumentation.grpc as otel_grpc # type: ignore[import-not-found]
428 endpoint_attrs = _extract_endpoint_attributes(client_options)
429 request_hook = _make_grpc_client_request_hook(endpoint_attrs)
431 raw_interceptors = otel_grpc.aio_client_interceptors(
432 tracer_provider=_get_tracer_provider(client_options),
433 request_hook=request_hook,
434 response_hook=_grpc_client_response_hook,
435 )
436 interceptors = (
437 list(raw_interceptors)
438 if isinstance(raw_interceptors, (list, tuple))
439 else [raw_interceptors]
440 )
441 interceptors.append(_AsyncSuppressingClientInterceptor())
443 return interceptors
446_CHANNEL_INTERCEPTOR_MAPPING = (
447 ("intercept_unary_unary", "_unary_unary_interceptors"),
448 ("intercept_unary_stream", "_unary_stream_interceptors"),
449 ("intercept_stream_unary", "_stream_unary_interceptors"),
450 ("intercept_stream_stream", "_stream_stream_interceptors"),
451)
454def apply_otel_async_channel_interceptors(
455 channel: Any,
456 client_options: ClientOptions | dict[str, Any] | Sequence[Any] | None = None,
457) -> Any:
458 """Creates and attaches OpenTelemetry client interceptors to the front of an async gRPC channel.
460 In grpc.aio, channels maintain internal interceptor lists evaluated in index order.
461 To ensure Google Cloud SDK telemetry wraps outside downstream interceptors and
462 suppresses ambient generic auto-instrumentation, our interceptors are prepended
463 to the front of the channel's interceptor lists.
465 Args:
466 channel: The async gRPC channel to intercept.
467 client_options (Optional[Union[ClientOptions, dict[str, Any], Sequence[Any]]]): The client
468 options used to resolve feature flags and tracer provider, or a sequence of
469 pre-resolved interceptors.
471 Returns:
472 The channel with interceptors prepended to its internal lists,
473 or the original channel if tracing is disabled or not installed.
474 """
475 otel_interceptors: Sequence[Any] | None
476 if isinstance(client_options, (ClientOptions, dict)) or client_options is None:
477 otel_interceptors = get_otel_async_interceptor(client_options)
478 elif isinstance(client_options, Sequence):
479 otel_interceptors = client_options
480 else:
481 otel_interceptors = None
483 if not otel_interceptors:
484 return channel
486 interceptor_list = list(otel_interceptors)
488 # Section 1: Typed interceptors - prepend matching interceptors to each
489 # corresponding channel list (unary-unary, unary-stream, etc.) in order.
490 # See note in the docstring about why prepending is important here.
491 for method_name, attr_name in _CHANNEL_INTERCEPTOR_MAPPING:
492 target_list = getattr(channel, attr_name, None)
493 if isinstance(target_list, list):
494 matching = [
495 i
496 for i in interceptor_list
497 if hasattr(i, method_name) and i not in target_list
498 ]
499 if matching:
500 target_list[:0] = matching
502 # Section 2: Untyped/generic fallback - prepend any remaining interceptors
503 # lacking standard method names directly to the unary-unary list.
504 unary_list = getattr(channel, "_unary_unary_interceptors", None)
505 if isinstance(unary_list, list):
506 standard_methods = tuple(method for method, _ in _CHANNEL_INTERCEPTOR_MAPPING)
507 fallback_items = [
508 i
509 for i in interceptor_list
510 if not any(hasattr(i, m) for m in standard_methods) and i not in unary_list
511 ]
512 if fallback_items:
513 unary_list[:0] = fallback_items
515 return channel
518_TRACE_CONTEXT_PROPAGATOR: Any = None
521class _TraceContext:
522 """Context manager for tracing an HTTP wire request with OpenTelemetry.
524 Manages span creation, semantic attribute attachment, W3C traceparent injection,
525 and automatic error recording on failure without suppressing caller exceptions.
527 Supports two calling conventions:
528 - Pass a single `request` object (such as `requests.PreparedRequest`).
529 - Pass explicit keyword arguments (`method`, `url`, `headers`, `body`, `client_options`).
530 """
532 @staticmethod
533 def _get_trace_context_propagator() -> Any:
534 global _TRACE_CONTEXT_PROPAGATOR
535 if _TRACE_CONTEXT_PROPAGATOR is None:
536 from opentelemetry.trace.propagation.tracecontext import ( # type: ignore[import-not-found]
537 TraceContextTextMapPropagator,
538 )
540 _TRACE_CONTEXT_PROPAGATOR = TraceContextTextMapPropagator()
541 return _TRACE_CONTEXT_PROPAGATOR
543 # The HTTP tracing context manager deliberately supports two distinct invocation styles:
544 # 1. Bundled Request Object: `trace_http_request(request, ...)`
545 # Used when callers already possess an HTTP request instance (such as
546 # requests.PreparedRequest or urllib.request.Request) with `.method`, `.url`, etc.
547 # 2. Unpacked Keyword Arguments: `trace_http_request(method=..., url=..., headers=..., body=...)`
548 # Used by generated GAPIC REST transports (_shared_macros.j2).
549 # In GAPIC templates, requests are assembled from local strings and dictionaries before
550 # hitting the session. Supporting keyword arguments avoids the CPU and memory overhead
551 # of instantiating a throwaway dummy request object on every single RPC execution.
552 @staticmethod
553 def _build_http_span_attributes(
554 request: Any = None,
555 *,
556 method: str | None = None,
557 url: str | None = None,
558 url_template: str | None = None,
559 headers: dict[str, Any] | None = None,
560 body: Any = None,
561 client_options: ClientOptions | dict[str, Any] | None = None,
562 ) -> tuple[str, dict[str, Any], Any]:
563 """Extract span name, semantic attributes dictionary, and resolved headers.
565 Supports two calling conventions:
566 - Pass a single `request` object (such as `requests.PreparedRequest`).
567 - Pass explicit keyword arguments (`method`, `url`, `headers`, `body`, `client_options`).
569 Returns:
570 tuple[str, dict[str, Any], Any]: A tuple of (span_name, attributes, resolved_headers).
571 """
572 if request is not None:
573 resolved_method = getattr(request, "method", "HTTP") or "HTTP"
574 resolved_url = getattr(request, "url", "") or ""
575 resolved_headers = getattr(request, "headers", None)
576 resolved_body = getattr(request, "body", None)
577 else:
578 resolved_method = method or "HTTP"
579 resolved_url = url or ""
580 resolved_headers = headers
581 resolved_body = body
583 resolved_method = resolved_method.upper()
584 endpoint_attrs = _extract_endpoint_attributes(client_options)
586 server_address = endpoint_attrs.get("server.address")
587 server_port = endpoint_attrs.get("server.port")
588 if not server_address and resolved_url:
589 try:
590 parsed = urllib.parse.urlsplit(resolved_url)
591 server_address = parsed.hostname
592 if not server_port and parsed.port:
593 server_port = parsed.port
594 except Exception: # Fail-open on malformed URL parsing
595 pass
597 span_name = resolved_method
598 span_attributes: dict[str, Any] = {
599 "http.request.method": resolved_method,
600 "server.address": server_address or "",
601 "server.port": server_port or 443,
602 "url.domain": endpoint_attrs.get("url.domain", "googleapis.com"),
603 }
604 if url_template:
605 span_attributes["url.template"] = url_template
606 if resolved_url:
607 span_attributes["url.full"] = resolved_url
609 if resolved_body is not None and isinstance(resolved_body, (bytes, str)):
610 span_attributes["http.request.body.size"] = len(resolved_body)
612 return span_name, span_attributes, resolved_headers
614 def __init__(
615 self,
616 request: Any = None,
617 *,
618 method: str | None = None,
619 url: str | None = None,
620 url_template: str | None = None,
621 headers: dict[str, Any] | None = None,
622 body: Any = None,
623 client_options: ClientOptions | dict[str, Any] | None = None,
624 ):
625 self._request = request
626 self._method = method
627 self._url = url
628 self._url_template = url_template
629 self._headers = headers
630 self._body = body
631 self._client_options = client_options
632 self._span: Any = None
633 self._cm: Any = None
635 def record_response(self, response: Any) -> None:
636 """Record HTTP response attributes on the active span.
638 Args:
639 response (Any): The HTTP response object (e.g. requests.Response).
640 """
641 span = self._span
642 if span is None or not hasattr(span, "set_attribute"):
643 return
645 try:
646 from opentelemetry.trace.status import ( # type: ignore[import-not-found]
647 Status,
648 StatusCode,
649 )
651 status_code = getattr(
652 response, "status_code", getattr(response, "status", None)
653 )
654 if status_code is not None:
655 span.set_attribute("http.response.status_code", int(status_code))
656 if int(status_code) >= 400:
657 span.set_status(Status(StatusCode.ERROR))
658 else:
659 span.set_status(Status(StatusCode.OK))
661 headers = getattr(response, "headers", None)
662 if headers and "Content-Length" in headers:
663 try:
664 span.set_attribute(
665 "http.response.body.size", int(headers["Content-Length"])
666 )
667 except (ValueError, TypeError):
668 pass
669 elif hasattr(response, "_content") and response._content is not None:
670 try:
671 span.set_attribute(
672 "http.response.body.size", len(response._content)
673 )
674 except Exception:
675 pass
676 except Exception:
677 pass
679 # Alias for method name parity with Daniel's suggested convention
680 record_http_response = record_response
682 def record_error(self, exc: BaseException | None) -> None:
683 """Record supplemental Cloud Observability error attributes on the active span.
685 Upstream OpenTelemetry's ``start_as_current_span`` context manager automatically
686 records the exception event and marks the span status as ERROR when an unhandled
687 exception escapes (via its default ``record_exception=True`` and
688 ``set_status_on_exception=True`` settings).
690 This method supplements that base OpenTelemetry behavior by enriching the span
691 with semantic attributes required by Cloud Trace and the Google Cloud Observability
692 specification:
693 - ``error.type``: Low-cardinality status code string (e.g. "404") or exception class name.
694 - ``status.message``: Error description message.
696 Args:
697 exc (Optional[BaseException]): The exception raised during dispatch.
698 """
699 span = self._span
700 if span is None or exc is None:
701 return
703 try:
704 if hasattr(span, "set_attribute"):
705 status_code = getattr(exc, "code", None) or getattr(
706 exc, "status_code", None
707 )
708 if status_code:
709 span.set_attribute("error.type", str(status_code))
710 else:
711 span.set_attribute("error.type", exc.__class__.__name__)
712 msg = str(exc)
713 if msg:
714 span.set_attribute("status.message", msg)
715 except Exception: # Fail-open on error attribute extraction failure
716 pass
718 # Alias for method name parity with Daniel's suggested convention
719 record_http_error = record_error
721 def __enter__(self) -> "_TraceContext":
722 if not is_otel_capabilities_enabled(self._client_options):
723 return self
725 try:
726 from opentelemetry import trace
728 tracer_provider = _get_tracer_provider(self._client_options)
729 if tracer_provider is not None:
730 tracer = tracer_provider.get_tracer("google.api_core")
731 else:
732 tracer = trace.get_tracer("google.api_core")
734 span_name, span_attributes, resolved_headers = (
735 self._build_http_span_attributes(
736 self._request,
737 method=self._method,
738 url=self._url,
739 url_template=self._url_template,
740 headers=self._headers,
741 body=self._body,
742 client_options=self._client_options,
743 )
744 )
746 self._cm = tracer.start_as_current_span(
747 span_name,
748 kind=trace.SpanKind.CLIENT,
749 attributes=span_attributes,
750 )
751 self._span = self._cm.__enter__()
753 if resolved_headers is not None and hasattr(
754 resolved_headers, "__setitem__"
755 ):
756 try:
757 self._get_trace_context_propagator().inject(resolved_headers)
758 except Exception: # Fail-open on header injection failure
759 pass
761 return self
762 except Exception:
763 # Fail-open: telemetry failures must never disrupt core RPC execution
764 return self
766 def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
767 if self._cm is not None:
768 try:
769 if exc_val is not None and self._span is not None:
770 self.record_error(exc_val)
771 finally:
772 self._cm.__exit__(exc_type, exc_val, exc_tb)
773 # Always return None so caller exceptions are never suppressed
774 return None
777def trace_http_request(
778 request: Any = None,
779 *,
780 method: str | None = None,
781 url: str | None = None,
782 url_template: str | None = None,
783 headers: dict[str, Any] | None = None,
784 body: Any = None,
785 client_options: ClientOptions | dict[str, Any] | None = None,
786) -> _TraceContext:
787 """Context manager factory for tracing an HTTP wire request with OpenTelemetry.
789 Manages span creation, semantic attribute attachment, W3C traceparent injection,
790 and automatic error recording on failure without suppressing caller exceptions.
792 Supports two calling conventions:
793 1. Bundled Request Object: `trace_http_request(request, ...)`
794 Used when callers already possess an HTTP request instance (such as
795 `requests.PreparedRequest` or `urllib.request.Request`) with `.method`, `.url`, etc.
796 2. Unpacked Keyword Arguments: `trace_http_request(method=..., url=..., headers=..., body=...)`
797 Used by generated GAPIC REST transports (_shared_macros.j2).
798 In GAPIC templates, requests are assembled from local strings and dictionaries before
799 hitting the session. Supporting keyword arguments avoids the CPU and memory overhead
800 of instantiating a throwaway dummy request object on every single RPC execution.
802 Args:
803 request (Optional[Any]): Bundled HTTP request object with .method, .url, .headers.
804 method (Optional[str]): Explicit HTTP method (e.g. "GET", "POST").
805 url (Optional[str]): Fully qualified request URL.
806 url_template (Optional[str]): Parameterized path template for url.template.
807 headers (Optional[dict[str, Any]]): Mutable dictionary of request headers.
808 body (Optional[Any]): Request payload (used for body size calculation).
809 client_options (Optional[Union[ClientOptions, dict[str, Any]]]): Client options
810 containing custom endpoint and tracer provider configurations.
812 Returns:
813 _TraceContext: An active context manager managing the HTTP client span.
814 """
815 return _TraceContext(
816 request=request,
817 method=method,
818 url=url,
819 url_template=url_template,
820 headers=headers,
821 body=body,
822 client_options=client_options,
823 )