diff --git a/packages/google-api-core/google/api_core/_observability.py b/packages/google-api-core/google/api_core/_observability.py index f101cec28f5c..07cc352cd31c 100644 --- a/packages/google-api-core/google/api_core/_observability.py +++ b/packages/google-api-core/google/api_core/_observability.py @@ -18,6 +18,7 @@ from __future__ import annotations +import urllib.parse from typing import TYPE_CHECKING, Any, Callable, Sequence from google.api_core import _feature_gating_helpers @@ -25,7 +26,7 @@ if TYPE_CHECKING: # flake8: grpc, trace, and ClientInterceptor are imported only for static analysis and type annotations - # The `# noqa: F401` comment avoids flake8 "imported but not used" errors. + # The 'noqa: F401' comment avoids flake8 "imported but not used" errors. import grpc # noqa: F401 import opentelemetry.trace # noqa: F401 @@ -64,6 +65,131 @@ def is_otel_capabilities_enabled( return False +def _extract_endpoint_attributes( + client_options: ClientOptions | dict[str, Any] | None = None, +) -> dict[str, Any]: + """Extracts server.address, server.port (if non-default), and url.domain from client options if present. + + Args: + client_options: The client options object or dictionary. + + Returns: + dict[str, Any]: A dictionary containing url.domain and, if an api_endpoint is configured, + server.address and non-default server.port. + """ + attrs: dict[str, Any] = {} + endpoint = None + universe_domain = None + + if isinstance(client_options, dict): + endpoint = client_options.get("api_endpoint") + universe_domain = client_options.get("universe_domain") + elif client_options is not None: + endpoint = getattr(client_options, "api_endpoint", None) + universe_domain = getattr(client_options, "universe_domain", None) + + attrs["url.domain"] = universe_domain or "googleapis.com" + + if endpoint and isinstance(endpoint, str): + target = endpoint if "//" in endpoint else f"//{endpoint}" + parsed = urllib.parse.urlsplit(target) + if parsed.hostname: + attrs["server.address"] = parsed.hostname + if parsed.port: + scheme = parsed.scheme.lower() + is_default_port = (parsed.port == 443 and scheme in ("https", "")) or ( + parsed.port == 80 and scheme == "http" + ) + if not is_default_port: + attrs["server.port"] = parsed.port + return attrs + + +def _extract_grpc_request_attributes(request: Any) -> dict[str, Any]: + """Extracts Google Cloud T4 semantic and resource attributes from a gRPC request object. + + Args: + request: The gRPC request object. + + Returns: + dict[str, Any]: A dictionary of semantic attributes. + """ + attrs: dict[str, Any] = { + "rpc.system.name": "grpc", + } + if request is None: + return attrs + + resend_count = getattr(request, "resend_count", None) + if isinstance(resend_count, int) and resend_count > 0: + attrs["gcp.grpc.resend_count"] = resend_count + + return attrs + + +def _make_grpc_client_request_hook( + endpoint_attrs: dict[str, Any] | None = None, +) -> Callable[[Any, Any], None]: + """Creates an OpenTelemetry gRPC client request hook with optional endpoint attributes. + + Args: + endpoint_attrs: Optional static endpoint attributes to attach to every span. + + Returns: + Callable[[Any, Any], None]: The request hook callback. + """ + static_attrs = dict(endpoint_attrs) if endpoint_attrs else {} + + def client_request_hook(span: Any, request: Any) -> None: + if span is None or not getattr(span, "is_recording", lambda: True)(): + return + + # Upstream opentelemetry-instrumentation-grpc names spans with a leading slash + # (e.g. "/package.Service/Method") and sets only the short name on rpc.method. + # Normalize span.name and rpc.method to the fully-qualified name without leading slash. + span_name = getattr(span, "name", None) + clean_method_name = None + if isinstance(span_name, str) and span_name.startswith("/"): + clean_method_name = span_name.lstrip("/") + if hasattr(span, "update_name"): + span.update_name(clean_method_name) + + # Remove duplicate legacy rpc.system attribute set by stock instrumentation + # in favor of modern rpc.system.name ("grpc") per PRD changelog. + span_attributes = getattr(span, "_attributes", None) + if span_attributes is not None: + pop_fn = getattr(span_attributes, "pop", None) + if callable(pop_fn): + pop_fn("rpc.system", None) + + attrs = _extract_grpc_request_attributes(request) + if clean_method_name: + attrs["rpc.method"] = clean_method_name + if static_attrs: + attrs.update(static_attrs) + for key, value in attrs.items(): + span.set_attribute(key, value) + + return client_request_hook + + +_grpc_client_request_hook = _make_grpc_client_request_hook() + + +def _grpc_client_response_hook(span: Any, response: Any) -> None: + """OpenTelemetry gRPC client response hook to record response status code. + + Note: Upstream OpenTelemetry gRPC instrumentation only invokes this response_hook + on successful RPC invocations. Failed RPCs raise an exception before this hook is reached. + + Args: + span: The OpenTelemetry span. + response: The gRPC response object or details. + """ + if span is not None and hasattr(span, "set_attribute"): + span.set_attribute("rpc.response.status_code", "OK") + + def _get_tracer_provider( client_options: ClientOptions | dict[str, Any] | None = None, ) -> opentelemetry.trace.TracerProvider | None: @@ -101,8 +227,13 @@ def get_otel_interceptor( import opentelemetry.instrumentation.grpc as otel_grpc # type: ignore[import-not-found] + endpoint_attrs = _extract_endpoint_attributes(client_options) + request_hook = _make_grpc_client_request_hook(endpoint_attrs) + interceptor: ClientInterceptor = otel_grpc.client_interceptor( - tracer_provider=_get_tracer_provider(client_options) + tracer_provider=_get_tracer_provider(client_options), + request_hook=request_hook, + response_hook=_grpc_client_response_hook, ) def otel_interceptor(channel: grpc.Channel) -> grpc.Channel: @@ -130,6 +261,11 @@ def get_otel_async_interceptor( # Ignored by mypy: Optional dependency only loaded if early-return is skipped import opentelemetry.instrumentation.grpc as otel_grpc # type: ignore[import-not-found] + endpoint_attrs = _extract_endpoint_attributes(client_options) + request_hook = _make_grpc_client_request_hook(endpoint_attrs) + return otel_grpc.aio_client_interceptors( - tracer_provider=_get_tracer_provider(client_options) + tracer_provider=_get_tracer_provider(client_options), + request_hook=request_hook, + response_hook=_grpc_client_response_hook, ) diff --git a/packages/google-api-core/tests/unit/test_observability.py b/packages/google-api-core/tests/unit/test_observability.py index 8e8964e66264..635505dea8c3 100644 --- a/packages/google-api-core/tests/unit/test_observability.py +++ b/packages/google-api-core/tests/unit/test_observability.py @@ -13,6 +13,7 @@ # limitations under the License. import sys +import types from unittest import mock import pytest @@ -162,8 +163,15 @@ def test_get_otel_interceptor_enabled(monkeypatch): assert callable(interceptor) mock_otel_grpc.client_interceptor.assert_called_once_with( - tracer_provider=mock_tracer_provider + tracer_provider=mock_tracer_provider, + request_hook=mock.ANY, + response_hook=_observability._grpc_client_response_hook, ) + req_hook = mock_otel_grpc.client_interceptor.call_args[1]["request_hook"] + mock_span = mock.Mock() + mock_span.is_recording.return_value = True + req_hook(mock_span, None) + mock_span.set_attribute.assert_any_call("url.domain", "googleapis.com") result = interceptor(mock_raw_channel) assert result is mock_wrapped_channel @@ -251,5 +259,245 @@ def test_get_otel_async_interceptor_enabled(monkeypatch): result = _observability.get_otel_async_interceptor(client_options=options) assert result is mock_async_interceptors mock_otel_grpc.aio_client_interceptors.assert_called_once_with( - tracer_provider=mock_tracer_provider + tracer_provider=mock_tracer_provider, + request_hook=mock.ANY, + response_hook=_observability._grpc_client_response_hook, ) + + req_hook = mock_otel_grpc.aio_client_interceptors.call_args[1]["request_hook"] + mock_span = mock.Mock() + mock_span.is_recording.return_value = True + req_hook(mock_span, None) + mock_span.set_attribute.assert_any_call("url.domain", "googleapis.com") + + +@pytest.mark.parametrize( + "client_options,expected_attrs", + [ + (None, {"url.domain": "googleapis.com"}), + ({}, {"url.domain": "googleapis.com"}), + (ClientOptions(api_endpoint=None), {"url.domain": "googleapis.com"}), + ({"universe_domain": "myuniverse.com"}, {"url.domain": "myuniverse.com"}), + ( + ClientOptions(universe_domain="custom.domain"), + {"url.domain": "custom.domain"}, + ), + ( + {"api_endpoint": "secretmanager.googleapis.com"}, + { + "server.address": "secretmanager.googleapis.com", + "url.domain": "googleapis.com", + }, + ), + ( + {"api_endpoint": "secretmanager.googleapis.com:443"}, + { + "server.address": "secretmanager.googleapis.com", + "url.domain": "googleapis.com", + }, + ), + ( + {"api_endpoint": "https://secretmanager.googleapis.com:443"}, + { + "server.address": "secretmanager.googleapis.com", + "url.domain": "googleapis.com", + }, + ), + ( + {"api_endpoint": "http://localhost:80"}, + {"server.address": "localhost", "url.domain": "googleapis.com"}, + ), + ( + ClientOptions(api_endpoint="https://my-custom-host.com:8443/"), + { + "server.address": "my-custom-host.com", + "server.port": 8443, + "url.domain": "googleapis.com", + }, + ), + ( + ClientOptions(api_endpoint="http://[::1]:8080"), + { + "server.address": "::1", + "server.port": 8080, + "url.domain": "googleapis.com", + }, + ), + ( + ClientOptions(api_endpoint="http:///"), + {"url.domain": "googleapis.com"}, + ), + ], +) +def test_extract_endpoint_attributes(client_options, expected_attrs): + """Proves that _extract_endpoint_attributes correctly parses server.address, non-default server.port, and url.domain.""" + assert _observability._extract_endpoint_attributes(client_options) == expected_attrs + + +@pytest.mark.parametrize( + "req,expected_attrs", + [ + (None, {"rpc.system.name": "grpc"}), + (types.SimpleNamespace(), {"rpc.system.name": "grpc"}), + ( + types.SimpleNamespace(name="projects/p1/secrets/s1"), + {"rpc.system.name": "grpc"}, + ), + ( + types.SimpleNamespace(parent="projects/parent-p1"), + {"rpc.system.name": "grpc"}, + ), + ( + types.SimpleNamespace(name="projects/p1/secrets/s1", resend_count=2), + { + "rpc.system.name": "grpc", + "gcp.grpc.resend_count": 2, + }, + ), + ( + types.SimpleNamespace(resend_count=0), + {"rpc.system.name": "grpc"}, + ), + ], +) +def test_extract_grpc_request_attributes(req, expected_attrs): + """Proves that _extract_grpc_request_attributes extracts all T4 gRPC attributes.""" + assert _observability._extract_grpc_request_attributes(req) == expected_attrs + + +def test_grpc_client_request_hook(): + """Proves that _grpc_client_request_hook attaches extracted T4 attributes to recording spans, + normalizes span names, sets fully qualified rpc.method, and removes legacy rpc.system. + """ + # Non-recording span should not set attributes + mock_span_non_rec = mock.Mock() + mock_span_non_rec.is_recording.return_value = False + _observability._grpc_client_request_hook(mock_span_non_rec, mock.Mock()) + mock_span_non_rec.set_attribute.assert_not_called() + + # None span should safely return + _observability._grpc_client_request_hook(None, mock.Mock()) + + # Recording span with default hook, leading slash in span.name, and legacy rpc.system + mock_span_rec = mock.Mock() + mock_span_rec.is_recording.return_value = True + mock_span_rec.name = ( + "/google.cloud.secretmanager.v1.SecretManagerService/ListSecrets" + ) + mock_span_rec._attributes = {"rpc.system": "grpc"} + req = types.SimpleNamespace(name="projects/my-proj/secrets/s1", resend_count=1) + + _observability._grpc_client_request_hook(mock_span_rec, req) + + # Verify span name normalized and rpc.method set to fully qualified name + mock_span_rec.update_name.assert_called_once_with( + "google.cloud.secretmanager.v1.SecretManagerService/ListSecrets" + ) + mock_span_rec.set_attribute.assert_any_call( + "rpc.method", "google.cloud.secretmanager.v1.SecretManagerService/ListSecrets" + ) + + # Verify rpc.system.name set and legacy rpc.system popped + mock_span_rec.set_attribute.assert_any_call("rpc.system.name", "grpc") + assert "rpc.system" not in mock_span_rec._attributes + + mock_span_rec.set_attribute.assert_any_call("gcp.grpc.resend_count", 1) + + # Custom hook with endpoint attributes and already-clean span name + endpoint_hook = _observability._make_grpc_client_request_hook( + {"server.address": "custom.api.com", "server.port": 443} + ) + mock_span_custom = mock.Mock() + mock_span_custom.is_recording.return_value = True + mock_span_custom.name = "already_clean_name" + endpoint_hook(mock_span_custom, req) + mock_span_custom.set_attribute.assert_any_call("server.address", "custom.api.com") + mock_span_custom.set_attribute.assert_any_call("server.port", 443) + mock_span_custom.update_name.assert_not_called() + + +def test_get_otel_interceptor_with_api_endpoint(monkeypatch): + """Proves that get_otel_interceptor injects server.address, server.port, and url.domain when api_endpoint is set.""" + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true") + options = ClientOptions( + api_endpoint="secretmanager.googleapis.com:8443", + universe_domain="custom-domain.com", + ) + + mock_otel = mock.Mock() + mock_otel_grpc = mock_otel.instrumentation.grpc + monkeypatch.setitem(sys.modules, "opentelemetry", mock_otel) + monkeypatch.setitem( + sys.modules, "opentelemetry.instrumentation", mock_otel.instrumentation + ) + monkeypatch.setitem( + sys.modules, "opentelemetry.instrumentation.grpc", mock_otel_grpc + ) + + interceptor = _observability.get_otel_interceptor(client_options=options) + assert callable(interceptor) + + # Verify custom request hook was passed + args, kwargs = mock_otel_grpc.client_interceptor.call_args + req_hook = kwargs["request_hook"] + assert req_hook is not _observability._grpc_client_request_hook + assert kwargs["response_hook"] is _observability._grpc_client_response_hook + + # Test invoking the custom hook + mock_span = mock.Mock() + mock_span.is_recording.return_value = True + req_hook(mock_span, None) + mock_span.set_attribute.assert_any_call( + "server.address", "secretmanager.googleapis.com" + ) + mock_span.set_attribute.assert_any_call("server.port", 8443) + mock_span.set_attribute.assert_any_call("url.domain", "custom-domain.com") + + +def test_get_otel_async_interceptor_with_api_endpoint(monkeypatch): + """Proves that get_otel_async_interceptor injects server.address, server.port, and url.domain when api_endpoint is set.""" + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true") + options = ClientOptions( + api_endpoint="secretmanager.googleapis.com:8443", + universe_domain="custom-domain.com", + ) + + mock_otel = mock.Mock() + mock_otel_grpc = mock_otel.instrumentation.grpc + monkeypatch.setitem(sys.modules, "opentelemetry", mock_otel) + monkeypatch.setitem( + sys.modules, "opentelemetry.instrumentation", mock_otel.instrumentation + ) + monkeypatch.setitem( + sys.modules, "opentelemetry.instrumentation.grpc", mock_otel_grpc + ) + + result = _observability.get_otel_async_interceptor(client_options=options) + assert result is not None + + args, kwargs = mock_otel_grpc.aio_client_interceptors.call_args + req_hook = kwargs["request_hook"] + assert req_hook is not _observability._grpc_client_request_hook + assert kwargs["response_hook"] is _observability._grpc_client_response_hook + + mock_span = mock.Mock() + mock_span.is_recording.return_value = True + req_hook(mock_span, None) + mock_span.set_attribute.assert_any_call( + "server.address", "secretmanager.googleapis.com" + ) + mock_span.set_attribute.assert_any_call("server.port", 8443) + mock_span.set_attribute.assert_any_call("url.domain", "custom-domain.com") + + +def test_grpc_client_response_hook_success(): + """Proves that _grpc_client_response_hook sets rpc.response.status_code to 'OK' on success.""" + mock_span = mock.Mock() + _observability._grpc_client_response_hook(mock_span, mock.Mock()) + mock_span.set_attribute.assert_called_once_with("rpc.response.status_code", "OK") + + +def test_grpc_client_response_hook_none_or_missing_set_attribute(): + """Proves that _grpc_client_response_hook handles None or invalid span gracefully.""" + _observability._grpc_client_response_hook(None, mock.Mock()) + _observability._grpc_client_response_hook(object(), mock.Mock())