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 default_host: str | None = None,
80) -> dict[str, Any]:
81 """Extracts server.address, server.port (if non-default), and url.domain from client options if present.
83 Args:
84 client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options
85 object or dictionary.
86 default_host (Optional[str]): Fallback transport host when ``api_endpoint`` is not
87 set on ``client_options``.
89 Returns:
90 dict[str, Any]: A dictionary containing url.domain and, if an api_endpoint or default_host
91 is configured, server.address and non-default server.port.
92 """
93 attrs: dict[str, Any] = {}
94 endpoint = None
95 universe_domain = None
97 if isinstance(client_options, dict):
98 endpoint = client_options.get("api_endpoint") or default_host
99 universe_domain = client_options.get("universe_domain")
100 elif client_options is not None:
101 endpoint = getattr(client_options, "api_endpoint", None) or default_host
102 universe_domain = getattr(client_options, "universe_domain", None)
103 else:
104 endpoint = default_host
106 attrs["url.domain"] = universe_domain or "googleapis.com"
108 if endpoint and isinstance(endpoint, str):
109 target = endpoint if "//" in endpoint else f"//{endpoint}"
110 parsed = None
111 hostname = None
112 port = None
113 try:
114 parsed = urllib.parse.urlsplit(target)
115 hostname = parsed.hostname
116 port = parsed.port
117 except ValueError:
118 pass
120 if hostname:
121 attrs["server.address"] = hostname
122 if port and parsed:
123 scheme = parsed.scheme.lower()
124 is_default_port = (port == 443 and scheme in ("https", "")) or (
125 port == 80 and scheme == "http"
126 )
127 if not is_default_port:
128 attrs["server.port"] = port
129 return attrs
132def _make_grpc_client_request_hook(
133 endpoint_attrs: dict[str, Any] | None = None,
134) -> Callable[[Any, Any], None]:
135 """Creates an OpenTelemetry gRPC client request hook with optional endpoint attributes.
137 Args:
138 endpoint_attrs (Optional[dict[str, Any]]): Optional static endpoint attributes to attach
139 to every span.
141 Returns:
142 Callable[[Any, Any], None]: The request hook callback.
143 """
144 static_attrs = dict(endpoint_attrs) if endpoint_attrs else {}
146 def client_request_hook(span: Any, request: Any) -> None:
147 if span is None or not getattr(span, "is_recording", lambda: True)():
148 return
150 # Upstream opentelemetry-instrumentation-grpc may format span names with a
151 # leading slash (e.g. "/package.Service/Method"). Normalize the span name
152 # and ensure rpc.method is always captured as the clean, fully-qualified name.
153 span_name = getattr(span, "name", None)
154 if isinstance(span_name, str) and span_name:
155 clean_method_name = span_name.lstrip("/")
156 if span_name.startswith("/") and hasattr(span, "update_name"):
157 span.update_name(clean_method_name)
158 span.set_attribute("rpc.method", clean_method_name)
160 attrs: dict[str, Any] = {
161 "rpc.system.name": "grpc",
162 }
163 if static_attrs:
164 attrs.update(static_attrs)
165 for key, value in attrs.items():
166 span.set_attribute(key, value)
168 return client_request_hook
171_grpc_client_request_hook = _make_grpc_client_request_hook()
174def _grpc_client_response_hook(span: Any, response: Any) -> None:
175 """OpenTelemetry gRPC client response hook to record successful response status.
177 Upstream ``opentelemetry-instrumentation-grpc`` sets the integer status code
178 ``rpc.grpc.status_code`` (e.g. 0), but does not record the modern string status
179 ``rpc.response.status_code`` (e.g. "OK") required by Cloud Trace and current
180 OpenTelemetry semantic conventions (v1.27.0+).
182 This hook enriches successful RPC attempt spans with ``rpc.response.status_code = "OK"``.
183 Errors and non-OK statuses are handled at the Tier 3 method span layer or upstream.
185 Upstream handles synchronous and asynchronous invocations differently:
186 - **Synchronous gRPC**: Upstream only invokes the response hook when an RPC call
187 succeeds. On failure, the hook is bypassed entirely.
188 - **Asynchronous gRPC**: Upstream invokes the response hook unconditionally for
189 both successes and failures (passing exception details on error). However, it
190 always marks ``span.status`` with an error status before calling the hook.
192 Because of this disparity, this hook checks ``span.status`` to guard against
193 async failure callbacks while allowing synchronous and successful asynchronous
194 calls to be marked "OK".
196 Note:
197 If upstream ``opentelemetry-instrumentation-grpc`` adds native support for
198 modern ``rpc.response.status_code`` in future releases, this hook can be retired.
200 Args:
201 span (Optional[Any]): The OpenTelemetry span.
202 response (Any): The gRPC response object or details.
203 """
204 if span is None or not getattr(span, "is_recording", lambda: False)():
205 return
207 # Guard against upstream async calls that invoke this hook on failures.
208 status = getattr(span, "status", None)
209 status_code = getattr(status, "status_code", None)
210 if (
211 getattr(status_code, "name", None) == "ERROR"
212 or getattr(status_code, "value", None) == 2
213 ):
214 return
216 span.set_attribute("rpc.response.status_code", "OK")
219def _get_tracer_provider(
220 client_options: ClientOptions | dict[str, Any] | None = None,
221) -> opentelemetry.trace.TracerProvider | None:
222 """Extracts the OpenTelemetry tracer provider from client options if present.
224 Args:
225 client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options
226 object or dictionary.
228 Returns:
229 opentelemetry.trace.TracerProvider | None: The tracer provider if present,
230 None otherwise.
231 """
232 if isinstance(client_options, dict):
233 return client_options.get(_TRACER_PROVIDER)
234 elif client_options is not None:
235 return getattr(client_options, _TRACER_PROVIDER, None)
236 return None
239try:
240 # flake8: `grpc` imported conditionally for runtime interceptor base classes
241 import grpc # noqa: F811
242 from grpc import (
243 StreamStreamClientInterceptor as _SyncStreamStreamClientInterceptor,
244 )
245 from grpc import (
246 StreamUnaryClientInterceptor as _SyncStreamUnaryClientInterceptor,
247 )
248 from grpc import (
249 UnaryStreamClientInterceptor as _SyncUnaryStreamClientInterceptor,
250 )
251 from grpc import (
252 UnaryUnaryClientInterceptor as _SyncUnaryUnaryClientInterceptor,
253 )
254 from grpc.aio import (
255 StreamStreamClientInterceptor as _AsyncStreamStreamClientInterceptor,
256 )
257 from grpc.aio import (
258 StreamUnaryClientInterceptor as _AsyncStreamUnaryClientInterceptor,
259 )
260 from grpc.aio import (
261 UnaryStreamClientInterceptor as _AsyncUnaryStreamClientInterceptor,
262 )
263 from grpc.aio import (
264 UnaryUnaryClientInterceptor as _AsyncUnaryUnaryClientInterceptor,
265 )
266except ImportError: # pragma: NO COVER
267 # mypy: Fallbacks: grpc sentinel when optional grpc is not installed in the environment.
268 # and distinct empty classes avoid duplicate base class 'object' TypeError at runtime.
269 grpc = None # type: ignore[assignment]
271 class _SyncUnaryUnaryClientInterceptor: # type: ignore[no-redef]
272 __slots__ = ()
274 class _SyncUnaryStreamClientInterceptor: # type: ignore[no-redef]
275 __slots__ = ()
277 class _SyncStreamUnaryClientInterceptor: # type: ignore[no-redef]
278 __slots__ = ()
280 class _SyncStreamStreamClientInterceptor: # type: ignore[no-redef]
281 __slots__ = ()
283 class _AsyncUnaryUnaryClientInterceptor: # type: ignore[no-redef]
284 __slots__ = ()
286 class _AsyncUnaryStreamClientInterceptor: # type: ignore[no-redef]
287 __slots__ = ()
289 class _AsyncStreamUnaryClientInterceptor: # type: ignore[no-redef]
290 __slots__ = ()
292 class _AsyncStreamStreamClientInterceptor: # type: ignore[no-redef]
293 __slots__ = ()
296class _SuppressingClientInterceptor(
297 _SyncUnaryUnaryClientInterceptor,
298 _SyncUnaryStreamClientInterceptor,
299 _SyncStreamUnaryClientInterceptor,
300 _SyncStreamStreamClientInterceptor,
301):
302 """Client interceptor that suppresses redundant downstream generic auto-instrumentation spans.
304 When users enable global OpenTelemetry gRPC auto-instrumentation (e.g. GrpcInstrumentorClient),
305 the underlying channel is wrapped with generic interceptors that emit bare-bones spans.
306 This interceptor wraps downstream calls in OpenTelemetry's native `suppress_instrumentation`
307 context manager so that our feature-rich Google Cloud SDK T4 span is emitted while
308 redundant generic downstream spans are bypassed.
309 """
311 __slots__ = ()
313 def intercept_unary_unary(
314 self, continuation: Any, client_call_details: Any, request: Any
315 ) -> Any:
316 with _suppress_instrumentation():
317 return continuation(client_call_details, request)
319 def intercept_unary_stream(
320 self, continuation: Any, client_call_details: Any, request: Any
321 ) -> Any:
322 with _suppress_instrumentation():
323 return continuation(client_call_details, request)
325 def intercept_stream_unary(
326 self, continuation: Any, client_call_details: Any, request_iterator: Any
327 ) -> Any:
328 with _suppress_instrumentation():
329 return continuation(client_call_details, request_iterator)
331 def intercept_stream_stream(
332 self, continuation: Any, client_call_details: Any, request_iterator: Any
333 ) -> Any:
334 with _suppress_instrumentation():
335 return continuation(client_call_details, request_iterator)
338class _AsyncSuppressingClientInterceptor(
339 _AsyncUnaryUnaryClientInterceptor,
340 _AsyncUnaryStreamClientInterceptor,
341 _AsyncStreamUnaryClientInterceptor,
342 _AsyncStreamStreamClientInterceptor,
343):
344 """AsyncIO client interceptor that suppresses redundant downstream generic auto-instrumentation spans."""
346 __slots__ = ()
348 async def intercept_unary_unary(
349 self, continuation: Any, client_call_details: Any, request: Any
350 ) -> Any:
351 with _suppress_instrumentation():
352 return await continuation(client_call_details, request)
354 async def intercept_unary_stream(
355 self, continuation: Any, client_call_details: Any, request: Any
356 ) -> Any:
357 with _suppress_instrumentation():
358 return await continuation(client_call_details, request)
360 async def intercept_stream_unary(
361 self, continuation: Any, client_call_details: Any, request_iterator: Any
362 ) -> Any:
363 with _suppress_instrumentation():
364 return await continuation(client_call_details, request_iterator)
366 async def intercept_stream_stream(
367 self, continuation: Any, client_call_details: Any, request_iterator: Any
368 ) -> Any:
369 with _suppress_instrumentation():
370 return await continuation(client_call_details, request_iterator)
373def get_otel_interceptor(
374 client_options: ClientOptions | dict[str, Any] | None = None,
375 default_host: str | None = None,
376) -> Callable[[grpc.Channel], grpc.Channel] | None:
377 """Returns an interceptor callable that wraps a sync gRPC channel with OpenTelemetry tracing.
379 Args:
380 client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options
381 object or dictionary used for feature gating and extracting the tracer provider.
382 default_host (Optional[str]): Fallback transport host when ``api_endpoint`` is not
383 set on ``client_options``.
385 Returns:
386 Callable[[grpc.Channel], grpc.Channel] | None: An interceptor callable if OpenTelemetry
387 tracing is enabled and installed, None otherwise.
388 """
389 if not is_otel_capabilities_enabled(client_options):
390 return None
392 import opentelemetry.instrumentation.grpc as otel_grpc # type: ignore[import-not-found]
394 endpoint_attrs = _extract_endpoint_attributes(
395 client_options, default_host=default_host
396 )
397 request_hook = _make_grpc_client_request_hook(endpoint_attrs)
399 interceptor: ClientInterceptor = otel_grpc.client_interceptor(
400 tracer_provider=_get_tracer_provider(client_options),
401 request_hook=request_hook,
402 response_hook=_grpc_client_response_hook,
403 )
404 suppressor = _SuppressingClientInterceptor()
406 def otel_interceptor(channel: grpc.Channel) -> grpc.Channel:
407 if grpc is not None:
408 try:
409 channel = grpc.intercept_channel(channel, suppressor)
410 except (AttributeError, TypeError):
411 # Fallback if grpc lacks intercept_channel or custom channel type rejects it.
412 pass
413 return otel_grpc.intercept_channel(channel, interceptor)
415 otel_interceptor._is_otel_interceptor = True # type: ignore[attr-defined]
416 return otel_interceptor
419def get_otel_async_interceptor(
420 client_options: ClientOptions | dict[str, Any] | None = None,
421 default_host: str | None = None,
422) -> Sequence[grpc.aio.ClientInterceptor] | None:
423 """Returns async gRPC client interceptors for OpenTelemetry tracing.
425 Args:
426 client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options
427 object or dictionary used for feature gating and extracting the tracer provider.
428 default_host (Optional[str]): Fallback transport host when ``api_endpoint`` is not
429 set on ``client_options``.
431 Returns:
432 Sequence[grpc.aio.ClientInterceptor] | None: Instantiated OpenTelemetry async
433 client interceptors if tracing is enabled and installed, None otherwise.
434 """
435 if not is_otel_capabilities_enabled(client_options):
436 return None
438 # Ignored by mypy: Optional dependency only loaded if early-return is skipped
439 import opentelemetry.instrumentation.grpc as otel_grpc # type: ignore[import-not-found]
441 endpoint_attrs = _extract_endpoint_attributes(
442 client_options, default_host=default_host
443 )
444 request_hook = _make_grpc_client_request_hook(endpoint_attrs)
446 raw_interceptors = otel_grpc.aio_client_interceptors(
447 tracer_provider=_get_tracer_provider(client_options),
448 request_hook=request_hook,
449 response_hook=_grpc_client_response_hook,
450 )
451 interceptors = (
452 list(raw_interceptors)
453 if isinstance(raw_interceptors, (list, tuple))
454 else [raw_interceptors]
455 )
456 interceptors.append(_AsyncSuppressingClientInterceptor())
458 return interceptors
461_CHANNEL_INTERCEPTOR_MAPPING = (
462 ("intercept_unary_unary", "_unary_unary_interceptors"),
463 ("intercept_unary_stream", "_unary_stream_interceptors"),
464 ("intercept_stream_unary", "_stream_unary_interceptors"),
465 ("intercept_stream_stream", "_stream_stream_interceptors"),
466)
469def apply_otel_async_channel_interceptors(
470 channel: Any,
471 client_options: ClientOptions | dict[str, Any] | Sequence[Any] | None = None,
472 default_host: str | None = None,
473) -> Any:
474 """Creates and attaches OpenTelemetry client interceptors to the front of an async gRPC channel.
476 In grpc.aio, channels maintain internal interceptor lists evaluated in index order.
477 To ensure Google Cloud SDK telemetry wraps outside downstream interceptors and
478 suppresses ambient generic auto-instrumentation, our interceptors are prepended
479 to the front of the channel's interceptor lists.
481 Args:
482 channel: The async gRPC channel to intercept.
483 client_options (Optional[Union[ClientOptions, dict[str, Any], Sequence[Any]]]): The client
484 options used to resolve feature flags and tracer provider, or a sequence of
485 pre-resolved interceptors.
486 default_host (Optional[str]): Fallback transport host when ``api_endpoint`` is not
487 set on ``client_options``.
489 Returns:
490 The channel with interceptors prepended to its internal lists,
491 or the original channel if tracing is disabled or not installed.
492 """
493 otel_interceptors: Sequence[Any] | None
494 if isinstance(client_options, (ClientOptions, dict)) or client_options is None:
495 otel_interceptors = get_otel_async_interceptor(
496 client_options, default_host=default_host
497 )
498 elif isinstance(client_options, Sequence):
499 otel_interceptors = client_options
500 else:
501 otel_interceptors = None
503 if not otel_interceptors:
504 return channel
506 interceptor_list = list(otel_interceptors)
508 # Section 1: Typed interceptors - prepend matching interceptors to each
509 # corresponding channel list (unary-unary, unary-stream, etc.) in order.
510 # See note in the docstring about why prepending is important here.
511 for method_name, attr_name in _CHANNEL_INTERCEPTOR_MAPPING:
512 target_list = getattr(channel, attr_name, None)
513 if isinstance(target_list, list):
514 matching = [
515 i
516 for i in interceptor_list
517 if hasattr(i, method_name) and i not in target_list
518 ]
519 if matching:
520 target_list[:0] = matching
522 # Section 2: Untyped/generic fallback - prepend any remaining interceptors
523 # lacking standard method names directly to the unary-unary list.
524 unary_list = getattr(channel, "_unary_unary_interceptors", None)
525 if isinstance(unary_list, list):
526 standard_methods = tuple(method for method, _ in _CHANNEL_INTERCEPTOR_MAPPING)
527 fallback_items = [
528 i
529 for i in interceptor_list
530 if not any(hasattr(i, m) for m in standard_methods) and i not in unary_list
531 ]
532 if fallback_items:
533 unary_list[:0] = fallback_items
535 return channel
538_TRACE_CONTEXT_PROPAGATOR: Any = None
541class _TraceContext:
542 """Context manager for tracing an HTTP wire request with OpenTelemetry.
544 Manages span creation, semantic attribute attachment, W3C traceparent injection,
545 and automatic error recording on failure without suppressing caller exceptions.
547 Supports two calling conventions:
548 - Pass a single `request` object (such as `requests.PreparedRequest`).
549 - Pass explicit keyword arguments (`method`, `url`, `headers`, `body`, `client_options`).
550 """
552 @staticmethod
553 def _get_trace_context_propagator() -> Any:
554 global _TRACE_CONTEXT_PROPAGATOR
555 if _TRACE_CONTEXT_PROPAGATOR is None:
556 from opentelemetry.trace.propagation.tracecontext import ( # type: ignore[import-not-found]
557 TraceContextTextMapPropagator,
558 )
560 _TRACE_CONTEXT_PROPAGATOR = TraceContextTextMapPropagator()
561 return _TRACE_CONTEXT_PROPAGATOR
563 # The HTTP tracing context manager deliberately supports two distinct invocation styles:
564 # 1. Bundled Request Object: `trace_http_request(request, ...)`
565 # Used when callers already possess an HTTP request instance (such as
566 # requests.PreparedRequest or urllib.request.Request) with `.method`, `.url`, etc.
567 # 2. Unpacked Keyword Arguments: `trace_http_request(method=..., url=..., headers=..., body=...)`
568 # Used by generated GAPIC REST transports (_shared_macros.j2).
569 # In GAPIC templates, requests are assembled from local strings and dictionaries before
570 # hitting the session. Supporting keyword arguments avoids the CPU and memory overhead
571 # of instantiating a throwaway dummy request object on every single RPC execution.
572 @staticmethod
573 def _build_http_span_attributes(
574 request: Any = None,
575 *,
576 method: str | None = None,
577 url: str | None = None,
578 url_template: str | None = None,
579 headers: dict[str, Any] | None = None,
580 body: Any = None,
581 client_options: ClientOptions | dict[str, Any] | None = None,
582 ) -> tuple[str, dict[str, Any], Any]:
583 """Extract span name, semantic attributes dictionary, and resolved headers.
585 Supports two calling conventions:
586 - Pass a single `request` object (such as `requests.PreparedRequest`).
587 - Pass explicit keyword arguments (`method`, `url`, `headers`, `body`, `client_options`).
589 Returns:
590 tuple[str, dict[str, Any], Any]: A tuple of (span_name, attributes, resolved_headers).
591 """
592 if request is not None:
593 resolved_method = getattr(request, "method", "HTTP") or "HTTP"
594 resolved_url = getattr(request, "url", "") or ""
595 resolved_headers = getattr(request, "headers", None)
596 resolved_body = getattr(request, "body", None)
597 else:
598 resolved_method = method or "HTTP"
599 resolved_url = url or ""
600 resolved_headers = headers
601 resolved_body = body
603 resolved_method = resolved_method.upper()
604 endpoint_attrs = _extract_endpoint_attributes(client_options)
606 server_address = endpoint_attrs.get("server.address")
607 server_port = endpoint_attrs.get("server.port")
608 if not server_address and resolved_url:
609 try:
610 parsed = urllib.parse.urlsplit(resolved_url)
611 server_address = parsed.hostname
612 if not server_port and parsed.port:
613 server_port = parsed.port
614 except Exception: # Fail-open on malformed URL parsing
615 pass
617 span_name = (
618 f"{resolved_method} {url_template}" if url_template else resolved_method
619 )
620 span_attributes: dict[str, Any] = {
621 "rpc.system.name": "http",
622 "http.request.method": resolved_method,
623 "server.address": server_address or "",
624 "server.port": server_port or 443,
625 "url.domain": endpoint_attrs.get("url.domain", "googleapis.com"),
626 }
627 if url_template:
628 span_attributes["url.template"] = url_template
629 if resolved_url:
630 span_attributes["url.full"] = resolved_url
632 if resolved_body is not None and isinstance(resolved_body, (bytes, str)):
633 span_attributes["http.request.body.size"] = len(resolved_body)
635 return span_name, span_attributes, resolved_headers
637 @staticmethod
638 def _extract_http_error_message(response: Any) -> str | None:
639 """Extract a human-readable error message from an HTTP error response."""
640 json_fn = getattr(response, "json", None)
641 if callable(json_fn):
642 try:
643 payload = json_fn()
644 if isinstance(payload, dict):
645 err_obj = payload.get("error")
646 if isinstance(err_obj, dict):
647 msg = err_obj.get("message")
648 if isinstance(msg, str) and msg:
649 return msg
650 elif isinstance(err_obj, str) and err_obj:
651 return err_obj
652 # Intentionally catch broad Exception so telemetry extraction
653 # fails open and never disrupts the caller's RPC flow.
654 except Exception:
655 pass
656 for attr in ("text", "reason"):
657 try:
658 val = getattr(response, attr, None)
659 if isinstance(val, str) and val.strip():
660 return val.strip()
661 # Intentionally catch broad Exception so telemetry extraction
662 # fails open and never disrupts the caller's RPC flow.
663 except Exception:
664 continue
665 return None
667 def __init__(
668 self,
669 request: Any = None,
670 *,
671 method: str | None = None,
672 url: str | None = None,
673 url_template: str | None = None,
674 headers: dict[str, Any] | None = None,
675 body: Any = None,
676 client_options: ClientOptions | dict[str, Any] | None = None,
677 ):
678 self._request = request
679 self._method = method
680 self._url = url
681 self._url_template = url_template
682 self._headers = headers
683 self._body = body
684 self._client_options = client_options
685 self._span: Any = None
686 self._parent_span: Any = None
687 self._cm: Any = None
689 def record_response(self, response: Any) -> None:
690 """Record HTTP response attributes on the active span.
692 Args:
693 response (Any): The HTTP response object (e.g. requests.Response).
694 """
695 span = self._span
696 if span is None or not hasattr(span, "set_attribute"):
697 return
699 try:
700 from opentelemetry.trace.status import ( # type: ignore[import-not-found]
701 Status,
702 StatusCode,
703 )
705 status_code = getattr(
706 response, "status_code", getattr(response, "status", None)
707 )
708 if status_code is not None:
709 code_int = int(status_code)
710 span.set_attribute("http.response.status_code", code_int)
711 if self._parent_span is not None:
712 self._parent_span.set_attribute(
713 "http.response.status_code", code_int
714 )
715 if code_int >= 400:
716 span.set_status(Status(StatusCode.ERROR))
717 span.set_attribute("error.type", str(code_int))
718 error_msg = self._extract_http_error_message(response)
719 if error_msg:
720 span.set_attribute("status.message", error_msg)
722 headers = getattr(response, "headers", None)
723 if headers and "Content-Length" in headers:
724 try:
725 span.set_attribute(
726 "http.response.body.size", int(headers["Content-Length"])
727 )
728 except (ValueError, TypeError):
729 pass
730 elif hasattr(response, "_content") and response._content is not None:
731 try:
732 span.set_attribute(
733 "http.response.body.size", len(response._content)
734 )
735 except Exception:
736 pass
737 except Exception:
738 pass
740 # Alias for method name parity with Daniel's suggested convention
741 record_http_response = record_response
743 def record_error(self, exc: BaseException | None) -> None:
744 """Record supplemental Cloud Observability error attributes on the active span.
746 Upstream OpenTelemetry's ``start_as_current_span`` context manager automatically
747 records the exception event and marks the span status as ERROR when an unhandled
748 exception escapes (via its default ``record_exception=True`` and
749 ``set_status_on_exception=True`` settings).
751 This method supplements that base OpenTelemetry behavior by enriching the span
752 with semantic attributes required by Cloud Trace and the Google Cloud Observability
753 specification:
754 - ``error.type``: Low-cardinality status code string (e.g. "404") or exception class name.
755 - ``status.message``: Error description message.
757 Args:
758 exc (Optional[BaseException]): The exception raised during dispatch.
759 """
760 span = self._span
761 if span is None or exc is None:
762 return
764 try:
765 if hasattr(span, "set_attribute"):
766 status_code = getattr(exc, "code", None) or getattr(
767 exc, "status_code", None
768 )
769 if status_code:
770 span.set_attribute("error.type", str(status_code))
771 else:
772 span.set_attribute("error.type", exc.__class__.__name__)
773 msg = str(exc)
774 if msg:
775 span.set_attribute("status.message", msg)
776 except Exception: # Fail-open on error attribute extraction failure
777 pass
779 # Alias for method name parity with Daniel's suggested convention
780 record_http_error = record_error
782 def __enter__(self) -> "_TraceContext":
783 if not is_otel_capabilities_enabled(self._client_options):
784 return self
786 try:
787 from opentelemetry import trace
789 tracer_provider = _get_tracer_provider(self._client_options)
790 if tracer_provider is not None:
791 tracer = tracer_provider.get_tracer("google.api_core")
792 else:
793 tracer = trace.get_tracer("google.api_core")
795 span_name, span_attributes, resolved_headers = (
796 self._build_http_span_attributes(
797 self._request,
798 method=self._method,
799 url=self._url,
800 url_template=self._url_template,
801 headers=self._headers,
802 body=self._body,
803 client_options=self._client_options,
804 )
805 )
807 parent_span = trace.get_current_span()
808 if (
809 parent_span is not None
810 and getattr(parent_span, "is_recording", lambda: False)() is True
811 and hasattr(parent_span, "set_attribute")
812 ):
813 self._parent_span = parent_span
814 if self._url_template:
815 try:
816 parent_attrs = getattr(parent_span, "attributes", None)
817 if parent_attrs is None or "url.template" not in parent_attrs:
818 parent_span.set_attribute(
819 "url.template", self._url_template
820 )
821 # Intentionally catch broad Exception so telemetry extraction
822 # fails open and never disrupts the caller's RPC flow.
823 except Exception:
824 pass
826 self._cm = tracer.start_as_current_span(
827 span_name,
828 kind=trace.SpanKind.CLIENT,
829 attributes=span_attributes,
830 )
831 self._span = self._cm.__enter__()
833 if resolved_headers is not None and hasattr(
834 resolved_headers, "__setitem__"
835 ):
836 try:
837 self._get_trace_context_propagator().inject(resolved_headers)
838 except Exception: # Fail-open on header injection failure
839 pass
841 return self
842 except Exception:
843 # Fail-open: telemetry failures must never disrupt core RPC execution
844 return self
846 def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
847 if self._cm is not None:
848 try:
849 if exc_val is not None and self._span is not None:
850 self.record_error(exc_val)
851 finally:
852 self._cm.__exit__(exc_type, exc_val, exc_tb)
853 # Always return None so caller exceptions are never suppressed
854 return None
857def trace_http_request(
858 request: Any = None,
859 *,
860 method: str | None = None,
861 url: str | None = None,
862 url_template: str | None = None,
863 headers: dict[str, Any] | None = None,
864 body: Any = None,
865 client_options: ClientOptions | dict[str, Any] | None = None,
866) -> _TraceContext:
867 """Context manager factory for tracing an HTTP wire request with OpenTelemetry.
869 Manages span creation, semantic attribute attachment, W3C traceparent injection,
870 and automatic error recording on failure without suppressing caller exceptions.
872 Supports two calling conventions:
873 1. Bundled Request Object: `trace_http_request(request, ...)`
874 Used when callers already possess an HTTP request instance (such as
875 `requests.PreparedRequest` or `urllib.request.Request`) with `.method`, `.url`, etc.
876 2. Unpacked Keyword Arguments: `trace_http_request(method=..., url=..., headers=..., body=...)`
877 Used by generated GAPIC REST transports (_shared_macros.j2).
878 In GAPIC templates, requests are assembled from local strings and dictionaries before
879 hitting the session. Supporting keyword arguments avoids the CPU and memory overhead
880 of instantiating a throwaway dummy request object on every single RPC execution.
882 Args:
883 request (Optional[Any]): Bundled HTTP request object with .method, .url, .headers.
884 method (Optional[str]): Explicit HTTP method (e.g. "GET", "POST").
885 url (Optional[str]): Fully qualified request URL.
886 url_template (Optional[str]): Parameterized path template for url.template.
887 headers (Optional[dict[str, Any]]): Mutable dictionary of request headers.
888 body (Optional[Any]): Request payload (used for body size calculation).
889 client_options (Optional[Union[ClientOptions, dict[str, Any]]]): Client options
890 containing custom endpoint and tracer provider configurations.
892 Returns:
893 _TraceContext: An active context manager managing the HTTP client span.
894 """
895 return _TraceContext(
896 request=request,
897 method=method,
898 url=url,
899 url_template=url_template,
900 headers=headers,
901 body=body,
902 client_options=client_options,
903 )