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 urllib.parse
22from typing import TYPE_CHECKING, Any, Callable, Sequence
24from google.api_core import _feature_gating_helpers
25from google.api_core.client_options import ClientOptions
27if TYPE_CHECKING:
28 # flake8: grpc, trace, and ClientInterceptor are imported only for static analysis and type annotations
29 # The 'noqa: F401' comment avoids flake8 "imported but not used" errors.
30 import grpc # noqa: F401
31 import opentelemetry.trace # noqa: F401
33 from google.api_core.grpc_helpers import ClientInterceptor # noqa: F401
35_TRACER_PROVIDER = "tracer_provider"
38def is_otel_capabilities_enabled(
39 client_options: ClientOptions | dict[str, Any] | None = None,
40 env_var: str = "GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED",
41) -> bool:
42 """Checks if OTel capabilities are enabled and installed.
44 Args:
45 client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options
46 object or dictionary.
47 env_var (str): The environment variable to check for enablement.
49 Returns:
50 bool: True if enabled and installed, False otherwise.
51 """
52 is_tracing_enabled = _feature_gating_helpers.resolve_feature_flags(
53 env_var=env_var,
54 feature_key=_TRACER_PROVIDER,
55 configuration=client_options,
56 )
58 if is_tracing_enabled:
59 try:
60 import opentelemetry.instrumentation.grpc as otel_grpc # type: ignore[import-not-found] # noqa: F401
62 return True
63 except ImportError:
64 pass
66 return False
69def _extract_endpoint_attributes(
70 client_options: ClientOptions | dict[str, Any] | None = None,
71) -> dict[str, Any]:
72 """Extracts server.address, server.port (if non-default), and url.domain from client options if present.
74 Args:
75 client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options
76 object or dictionary.
78 Returns:
79 dict[str, Any]: A dictionary containing url.domain and, if an api_endpoint is configured,
80 server.address and non-default server.port.
81 """
82 attrs: dict[str, Any] = {}
83 endpoint = None
84 universe_domain = None
86 if isinstance(client_options, dict):
87 endpoint = client_options.get("api_endpoint")
88 universe_domain = client_options.get("universe_domain")
89 elif client_options is not None:
90 endpoint = getattr(client_options, "api_endpoint", None)
91 universe_domain = getattr(client_options, "universe_domain", None)
93 attrs["url.domain"] = universe_domain or "googleapis.com"
95 if endpoint and isinstance(endpoint, str):
96 target = endpoint if "//" in endpoint else f"//{endpoint}"
97 parsed = None
98 hostname = None
99 port = None
100 try:
101 parsed = urllib.parse.urlsplit(target)
102 hostname = parsed.hostname
103 port = parsed.port
104 except ValueError:
105 pass
107 if hostname:
108 attrs["server.address"] = hostname
109 if port and parsed:
110 scheme = parsed.scheme.lower()
111 is_default_port = (port == 443 and scheme in ("https", "")) or (
112 port == 80 and scheme == "http"
113 )
114 if not is_default_port:
115 attrs["server.port"] = port
116 return attrs
119def _make_grpc_client_request_hook(
120 endpoint_attrs: dict[str, Any] | None = None,
121) -> Callable[[Any, Any], None]:
122 """Creates an OpenTelemetry gRPC client request hook with optional endpoint attributes.
124 Args:
125 endpoint_attrs (Optional[dict[str, Any]]): Optional static endpoint attributes to attach
126 to every span.
128 Returns:
129 Callable[[Any, Any], None]: The request hook callback.
130 """
131 static_attrs = dict(endpoint_attrs) if endpoint_attrs else {}
133 def client_request_hook(span: Any, request: Any) -> None:
134 if span is None or not getattr(span, "is_recording", lambda: True)():
135 return
137 # Upstream opentelemetry-instrumentation-grpc may format span names with a
138 # leading slash (e.g. "/package.Service/Method"). Normalize the span name
139 # and ensure rpc.method is always captured as the clean, fully-qualified name.
140 span_name = getattr(span, "name", None)
141 if isinstance(span_name, str) and span_name:
142 clean_method_name = span_name.lstrip("/")
143 if span_name.startswith("/") and hasattr(span, "update_name"):
144 span.update_name(clean_method_name)
145 span.set_attribute("rpc.method", clean_method_name)
147 attrs: dict[str, Any] = {
148 "rpc.system.name": "grpc",
149 }
150 if static_attrs:
151 attrs.update(static_attrs)
152 for key, value in attrs.items():
153 span.set_attribute(key, value)
155 return client_request_hook
158_grpc_client_request_hook = _make_grpc_client_request_hook()
161def _grpc_client_response_hook(span: Any, response: Any) -> None:
162 """OpenTelemetry gRPC client response hook to record successful response status.
164 Upstream ``opentelemetry-instrumentation-grpc`` sets the integer status code
165 ``rpc.grpc.status_code`` (e.g. 0), but does not record the modern string status
166 ``rpc.response.status_code`` (e.g. "OK") required by Cloud Trace and current
167 OpenTelemetry semantic conventions (v1.27.0+).
169 This hook enriches successful RPC attempt spans with ``rpc.response.status_code = "OK"``.
170 Errors and non-OK statuses are handled at the Tier 3 method span layer or upstream.
172 Upstream handles synchronous and asynchronous invocations differently:
173 - **Synchronous gRPC**: Upstream only invokes the response hook when an RPC call
174 succeeds. On failure, the hook is bypassed entirely.
175 - **Asynchronous gRPC**: Upstream invokes the response hook unconditionally for
176 both successes and failures (passing exception details on error). However, it
177 always marks ``span.status`` with an error status before calling the hook.
179 Because of this disparity, this hook checks ``span.status`` to guard against
180 async failure callbacks while allowing synchronous and successful asynchronous
181 calls to be marked "OK".
183 Note:
184 If upstream ``opentelemetry-instrumentation-grpc`` adds native support for
185 modern ``rpc.response.status_code`` in future releases, this hook can be retired.
187 Args:
188 span (Optional[Any]): The OpenTelemetry span.
189 response (Any): The gRPC response object or details.
190 """
191 if span is None or not getattr(span, "is_recording", lambda: False)():
192 return
194 # Guard against upstream async calls that invoke this hook on failures.
195 status = getattr(span, "status", None)
196 status_code = getattr(status, "status_code", None)
197 if (
198 getattr(status_code, "name", None) == "ERROR"
199 or getattr(status_code, "value", None) == 2
200 ):
201 return
203 span.set_attribute("rpc.response.status_code", "OK")
206def _get_tracer_provider(
207 client_options: ClientOptions | dict[str, Any] | None = None,
208) -> opentelemetry.trace.TracerProvider | None:
209 """Extracts the OpenTelemetry tracer provider from client options if present.
211 Args:
212 client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options
213 object or dictionary.
215 Returns:
216 opentelemetry.trace.TracerProvider | None: The tracer provider if present,
217 None otherwise.
218 """
219 if isinstance(client_options, dict):
220 return client_options.get(_TRACER_PROVIDER)
221 elif client_options is not None:
222 return getattr(client_options, _TRACER_PROVIDER, None)
223 return None
226def get_otel_interceptor(
227 client_options: ClientOptions | dict[str, Any] | None = None,
228) -> Callable[[grpc.Channel], grpc.Channel] | None:
229 """Returns an interceptor callable that wraps a sync gRPC channel with OpenTelemetry tracing.
231 Args:
232 client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options
233 object or dictionary used for feature gating and extracting the tracer provider.
235 Returns:
236 Callable[[grpc.Channel], grpc.Channel] | None: An interceptor callable if OpenTelemetry
237 tracing is enabled and installed, None otherwise.
238 """
239 if not is_otel_capabilities_enabled(client_options):
240 return None
242 import opentelemetry.instrumentation.grpc as otel_grpc # type: ignore[import-not-found]
244 endpoint_attrs = _extract_endpoint_attributes(client_options)
245 request_hook = _make_grpc_client_request_hook(endpoint_attrs)
247 interceptor: ClientInterceptor = otel_grpc.client_interceptor(
248 tracer_provider=_get_tracer_provider(client_options),
249 request_hook=request_hook,
250 response_hook=_grpc_client_response_hook,
251 )
253 def otel_interceptor(channel: grpc.Channel) -> grpc.Channel:
254 return otel_grpc.intercept_channel(channel, interceptor)
256 otel_interceptor._is_otel_interceptor = True # type: ignore[attr-defined]
257 return otel_interceptor
260def get_otel_async_interceptor(
261 client_options: ClientOptions | dict[str, Any] | None = None,
262) -> Sequence[grpc.aio.ClientInterceptor] | None:
263 """Returns async gRPC client interceptors for OpenTelemetry tracing.
265 Args:
266 client_options (Optional[Union[ClientOptions, dict[str, Any]]]): The client options
267 object or dictionary used for feature gating and extracting the tracer provider.
269 Returns:
270 Sequence[grpc.aio.ClientInterceptor] | None: Instantiated OpenTelemetry async
271 client interceptors if tracing is enabled and installed, None otherwise.
272 """
273 if not is_otel_capabilities_enabled(client_options):
274 return None
276 # Ignored by mypy: Optional dependency only loaded if early-return is skipped
277 import opentelemetry.instrumentation.grpc as otel_grpc # type: ignore[import-not-found]
279 endpoint_attrs = _extract_endpoint_attributes(client_options)
280 request_hook = _make_grpc_client_request_hook(endpoint_attrs)
282 return otel_grpc.aio_client_interceptors(
283 tracer_provider=_get_tracer_provider(client_options),
284 request_hook=request_hook,
285 response_hook=_grpc_client_response_hook,
286 )
289_TRACE_CONTEXT_PROPAGATOR: Any = None
292class _TraceContext:
293 """Context manager for tracing an HTTP wire request with OpenTelemetry.
295 Manages span creation, semantic attribute attachment, W3C traceparent injection,
296 and automatic error recording on failure without suppressing caller exceptions.
298 Supports two calling conventions:
299 - Pass a single `request` object (such as `requests.PreparedRequest`).
300 - Pass explicit keyword arguments (`method`, `url`, `headers`, `body`, `client_options`).
301 """
303 @staticmethod
304 def _get_trace_context_propagator() -> Any:
305 global _TRACE_CONTEXT_PROPAGATOR
306 if _TRACE_CONTEXT_PROPAGATOR is None:
307 from opentelemetry.trace.propagation.tracecontext import ( # type: ignore[import-not-found]
308 TraceContextTextMapPropagator,
309 )
311 _TRACE_CONTEXT_PROPAGATOR = TraceContextTextMapPropagator()
312 return _TRACE_CONTEXT_PROPAGATOR
314 # The HTTP tracing context manager deliberately supports two distinct invocation styles:
315 # 1. Bundled Request Object: `trace_http_request(request, ...)`
316 # Used when callers already possess an HTTP request instance (such as
317 # requests.PreparedRequest or urllib.request.Request) with `.method`, `.url`, etc.
318 # 2. Unpacked Keyword Arguments: `trace_http_request(method=..., url=..., headers=..., body=...)`
319 # Used by generated GAPIC REST transports (_shared_macros.j2).
320 # In GAPIC templates, requests are assembled from local strings and dictionaries before
321 # hitting the session. Supporting keyword arguments avoids the CPU and memory overhead
322 # of instantiating a throwaway dummy request object on every single RPC execution.
323 @staticmethod
324 def _build_http_span_attributes(
325 request: Any = None,
326 *,
327 method: str | None = None,
328 url: str | None = None,
329 url_template: str | None = None,
330 headers: dict[str, Any] | None = None,
331 body: Any = None,
332 client_options: ClientOptions | dict[str, Any] | None = None,
333 ) -> tuple[str, dict[str, Any], Any]:
334 """Extract span name, semantic attributes dictionary, and resolved headers.
336 Supports two calling conventions:
337 - Pass a single `request` object (such as `requests.PreparedRequest`).
338 - Pass explicit keyword arguments (`method`, `url`, `headers`, `body`, `client_options`).
340 Returns:
341 tuple[str, dict[str, Any], Any]: A tuple of (span_name, attributes, resolved_headers).
342 """
343 if request is not None:
344 resolved_method = getattr(request, "method", "HTTP") or "HTTP"
345 resolved_url = getattr(request, "url", "") or ""
346 resolved_headers = getattr(request, "headers", None)
347 resolved_body = getattr(request, "body", None)
348 else:
349 resolved_method = method or "HTTP"
350 resolved_url = url or ""
351 resolved_headers = headers
352 resolved_body = body
354 resolved_method = resolved_method.upper()
355 endpoint_attrs = _extract_endpoint_attributes(client_options)
357 server_address = endpoint_attrs.get("server.address")
358 server_port = endpoint_attrs.get("server.port")
359 if not server_address and resolved_url:
360 try:
361 parsed = urllib.parse.urlsplit(resolved_url)
362 server_address = parsed.hostname
363 if not server_port and parsed.port:
364 server_port = parsed.port
365 except Exception: # Fail-open on malformed URL parsing
366 pass
368 span_name = resolved_method
369 span_attributes: dict[str, Any] = {
370 "http.request.method": resolved_method,
371 "server.address": server_address or "",
372 "server.port": server_port or 443,
373 "url.domain": endpoint_attrs.get("url.domain", "googleapis.com"),
374 }
375 if url_template:
376 span_attributes["url.template"] = url_template
377 if resolved_url:
378 span_attributes["url.full"] = resolved_url
380 if resolved_body is not None and isinstance(resolved_body, (bytes, str)):
381 span_attributes["http.request.body.size"] = len(resolved_body)
383 return span_name, span_attributes, resolved_headers
385 def __init__(
386 self,
387 request: Any = None,
388 *,
389 method: str | None = None,
390 url: str | None = None,
391 url_template: str | None = None,
392 headers: dict[str, Any] | None = None,
393 body: Any = None,
394 client_options: ClientOptions | dict[str, Any] | None = None,
395 ):
396 self._request = request
397 self._method = method
398 self._url = url
399 self._url_template = url_template
400 self._headers = headers
401 self._body = body
402 self._client_options = client_options
403 self._span: Any = None
404 self._cm: Any = None
406 def record_response(self, response: Any) -> None:
407 """Record HTTP response attributes on the active span.
409 Args:
410 response (Any): The HTTP response object (e.g. requests.Response).
411 """
412 span = self._span
413 if span is None or not hasattr(span, "set_attribute"):
414 return
416 try:
417 from opentelemetry.trace.status import ( # type: ignore[import-not-found]
418 Status,
419 StatusCode,
420 )
422 status_code = getattr(
423 response, "status_code", getattr(response, "status", None)
424 )
425 if status_code is not None:
426 span.set_attribute("http.response.status_code", int(status_code))
427 if int(status_code) >= 400:
428 span.set_status(Status(StatusCode.ERROR))
429 else:
430 span.set_status(Status(StatusCode.OK))
432 headers = getattr(response, "headers", None)
433 if headers and "Content-Length" in headers:
434 try:
435 span.set_attribute(
436 "http.response.body.size", int(headers["Content-Length"])
437 )
438 except (ValueError, TypeError):
439 pass
440 elif hasattr(response, "_content") and response._content is not None:
441 try:
442 span.set_attribute(
443 "http.response.body.size", len(response._content)
444 )
445 except Exception:
446 pass
447 except Exception:
448 pass
450 # Alias for method name parity with Daniel's suggested convention
451 record_http_response = record_response
453 def record_error(self, exc: BaseException | None) -> None:
454 """Record supplemental Cloud Observability error attributes on the active span.
456 Upstream OpenTelemetry's ``start_as_current_span`` context manager automatically
457 records the exception event and marks the span status as ERROR when an unhandled
458 exception escapes (via its default ``record_exception=True`` and
459 ``set_status_on_exception=True`` settings).
461 This method supplements that base OpenTelemetry behavior by enriching the span
462 with semantic attributes required by Cloud Trace and the Google Cloud Observability
463 specification:
464 - ``error.type``: Low-cardinality status code string (e.g. "404") or exception class name.
465 - ``status.message``: Error description message.
467 Args:
468 exc (Optional[BaseException]): The exception raised during dispatch.
469 """
470 span = self._span
471 if span is None or exc is None:
472 return
474 try:
475 if hasattr(span, "set_attribute"):
476 status_code = getattr(exc, "code", None) or getattr(
477 exc, "status_code", None
478 )
479 if status_code:
480 span.set_attribute("error.type", str(status_code))
481 else:
482 span.set_attribute("error.type", exc.__class__.__name__)
483 msg = str(exc)
484 if msg:
485 span.set_attribute("status.message", msg)
486 except Exception: # Fail-open on error attribute extraction failure
487 pass
489 # Alias for method name parity with Daniel's suggested convention
490 record_http_error = record_error
492 def __enter__(self) -> "_TraceContext":
493 if not is_otel_capabilities_enabled(self._client_options):
494 return self
496 try:
497 from opentelemetry import trace
499 tracer_provider = _get_tracer_provider(self._client_options)
500 if tracer_provider is not None:
501 tracer = tracer_provider.get_tracer("google.api_core")
502 else:
503 tracer = trace.get_tracer("google.api_core")
505 span_name, span_attributes, resolved_headers = (
506 self._build_http_span_attributes(
507 self._request,
508 method=self._method,
509 url=self._url,
510 url_template=self._url_template,
511 headers=self._headers,
512 body=self._body,
513 client_options=self._client_options,
514 )
515 )
517 self._cm = tracer.start_as_current_span(
518 span_name,
519 kind=trace.SpanKind.CLIENT,
520 attributes=span_attributes,
521 )
522 self._span = self._cm.__enter__()
524 if resolved_headers is not None and hasattr(
525 resolved_headers, "__setitem__"
526 ):
527 try:
528 self._get_trace_context_propagator().inject(resolved_headers)
529 except Exception: # Fail-open on header injection failure
530 pass
532 return self
533 except Exception:
534 # Fail-open: telemetry failures must never disrupt core RPC execution
535 return self
537 def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
538 if self._cm is not None:
539 try:
540 if exc_val is not None and self._span is not None:
541 self.record_error(exc_val)
542 finally:
543 self._cm.__exit__(exc_type, exc_val, exc_tb)
544 # Always return None so caller exceptions are never suppressed
545 return None
548def trace_http_request(
549 request: Any = None,
550 *,
551 method: str | None = None,
552 url: str | None = None,
553 url_template: str | None = None,
554 headers: dict[str, Any] | None = None,
555 body: Any = None,
556 client_options: ClientOptions | dict[str, Any] | None = None,
557) -> _TraceContext:
558 """Context manager factory for tracing an HTTP wire request with OpenTelemetry.
560 Manages span creation, semantic attribute attachment, W3C traceparent injection,
561 and automatic error recording on failure without suppressing caller exceptions.
563 Supports two calling conventions:
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.
573 Args:
574 request (Optional[Any]): Bundled HTTP request object with .method, .url, .headers.
575 method (Optional[str]): Explicit HTTP method (e.g. "GET", "POST").
576 url (Optional[str]): Fully qualified request URL.
577 url_template (Optional[str]): Parameterized path template for url.template.
578 headers (Optional[dict[str, Any]]): Mutable dictionary of request headers.
579 body (Optional[Any]): Request payload (used for body size calculation).
580 client_options (Optional[Union[ClientOptions, dict[str, Any]]]): Client options
581 containing custom endpoint and tracer provider configurations.
583 Returns:
584 _TraceContext: An active context manager managing the HTTP client span.
585 """
586 return _TraceContext(
587 request=request,
588 method=method,
589 url=url,
590 url_template=url_template,
591 headers=headers,
592 body=body,
593 client_options=client_options,
594 )