Skip to content

feat(gapic): add OpenTelemetry T3 client method span wrapping in gapic_v1.method (D) #18274

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
chalmerlowe wants to merge 3 commits into from
Open

feat(gapic): add OpenTelemetry T3 client method span wrapping in gapic_v1.method (D) #18274

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
95ca42a
feat(gapic): add OpenTelemetry T3 client method span wrapping
chalmerlowe Sep 3, 2026
4cd28f0
chore(gapic): remove _is_tracing_supported dummy class variable
chalmerlowe Sep 3, 2026
9b46ef7
test(gapic): achieve 100% branch and statement coverage for OTel T3 m…
chalmerlowe Sep 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 37 additions & 1 deletion packages/google-api-core/google/api_core/gapic_v1/method.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
import functools
from typing import List, Tuple

from google.api_core import grpc_helpers
from google.api_core import _observability, grpc_helpers
from google.api_core.gapic_v1 import client_info
from google.api_core.timeout import TimeToDeadlineTimeout

Expand Down Expand Up @@ -186,6 +186,42 @@ def __call__(
if self._compression is not None:
kwargs["compression"] = compression

if _observability.is_otel_capabilities_enabled():
try:
from opentelemetry import trace

tracer = trace.get_tracer("google.api_core")

@daniel-sanche daniel-sanche Sep 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need to create this on every invocation? Can we cache it for each request? Or even use a singleton shared across all instances?

@daniel-sanche daniel-sanche Sep 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

how does this connect with the client's tracer provider? Is that coming later?

raw_method = getattr(self._target, "_method", None)

@daniel-sanche daniel-sanche Sep 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

have you tested this against a real client yet? IIRC, there are multiple layers of wrapping, so this may not be exposed the way you expect

@daniel-sanche daniel-sanche Sep 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I wonder if it would be possible to pass down a method name, instead of trying to extract it? The generator already knows it when calling _prep_wrapped_messages

if raw_method and isinstance(raw_method, (str, bytes)):
if isinstance(raw_method, bytes):
raw_method = raw_method.decode("utf-8")
method_str = raw_method.lstrip("/")
service, _, method = method_str.rpartition("/")
span_name = method_str
else:
service = "google.api_core"
method = getattr(self._target, "__name__", "call")
span_name = f"{service}/{method}"

@daniel-sanche daniel-sanche Sep 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the hot path that is called on every rpc. It seems like some of this would be doing the same (possibly slow) calculation on each invocation, right? Can we move that logic into the one-time init call?


@daniel-sanche daniel-sanche Sep 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

helper methods would be useful here

with tracer.start_as_current_span(
span_name,
kind=trace.SpanKind.CLIENT,
attributes={
"rpc.system": "grpc",

@daniel-sanche daniel-sanche Sep 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IIRC, this wrapper is also used by HTTP. So we should try to gate this for now

"rpc.service": service,
"rpc.method": method,
},
) as span:
try:
return wrapped_func(*args, **kwargs)

@daniel-sanche daniel-sanche Sep 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There are two places this function can call into wrapped_function. If errors line up the wrong way, it could hit both. We need to be extra careful to avoid double invocation here, because that would be a very serious bug

It might be better to call wrapped_func a single time at the end of the method, but use a no-op context manager instead of the tracer if we can't get one

@daniel-sanche daniel-sanche Sep 4, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Note that this span won't be very meaningful for streaming rpcs, because it just tracks the stream set-up, not any of the data flow. I remember asking Wes about streaming, and he said it's out of scope.

We should check with Blake if he wants to track stream init like this, or if we should avoid recording any data for streaming rpcs

except Exception as exc:
span.record_exception(exc)
span.set_status(trace.StatusCode.ERROR, str(exc))
raise
# If OpenTelemetry cannot be imported in the current environment, continue without tracing.
except ImportError: # pragma: NO COVER
pass
Comment on lines +189 to +223

@gemini-code-assist gemini-code-assist Bot Sep 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

Critical Reliability & Correctness Issues

  1. Resilience of Fallback Logic: Currently, only ImportError is caught. If any other exception occurs during OpenTelemetry setup, tracer retrieval, or span creation (e.g., AttributeError, TypeError, UnicodeDecodeError when decoding _method, or OTel initialization/configuration errors), it will propagate and crash the user's API call. Per the Repository Style Guide (Rule 3: Self-Contained Fallbacks), fallback logic must be resilient and self-contained, bypassing failures gracefully.
  2. Double Execution Risk: If we simply broaden the exception handler to except Exception:, any exception raised by wrapped_func (which is caught and re-raised by the inner except Exception as exc) would be caught by the outer except Exception and trigger a second execution of wrapped_func(*args, **kwargs). This is a critical bug that could cause non-idempotent RPCs to be executed twice.

Solution

We can use a state flag (func_called) to track whether wrapped_func has been invoked. This allows us to catch all exceptions during OTel setup/span creation and fallback gracefully, while ensuring that any exception raised by wrapped_func itself is propagated immediately without triggering a double execution.

        if _observability.is_otel_capabilities_enabled():
            func_called = False
            try:
                from opentelemetry import trace

                tracer = trace.get_tracer("google.api_core")
                raw_method = getattr(self._target, "_method", None)
                if raw_method and isinstance(raw_method, (str, bytes)):
                    if isinstance(raw_method, bytes):
                        raw_method = raw_method.decode("utf-8")
                    method_str = raw_method.lstrip("/")
                    service, _, method = method_str.rpartition("/")
                    span_name = method_str
                else:
                    service = "google.api_core"
                    method = getattr(self._target, "__name__", "call")
                    span_name = f"{service}/{method}"

                with tracer.start_as_current_span(
                    span_name,
                    kind=trace.SpanKind.CLIENT,
                    attributes={
                        "rpc.system": "grpc",
                        "rpc.service": service,
                        "rpc.method": method,
                    },
                ) as span:
                    try:
                        func_called = True
                        return wrapped_func(*args, **kwargs)
                    except Exception as exc:
                        span.record_exception(exc)
                        span.set_status(trace.StatusCode.ERROR, str(exc))
                        raise
            except Exception:
                if func_called:
                    raise
References
  1. Rule 3: Self-Contained Fallbacks - Fallback logic must be resilient and self-contained. Always wrap fallback configuration loading in try-except blocks to catch expected exceptions and bypass failures gracefully. (link)


return wrapped_func(*args, **kwargs)


Expand Down
312 changes: 312 additions & 0 deletions packages/google-api-core/tests/unit/gapic/test_method.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
# limitations under the License.

import datetime
import sys
from unittest import mock

import pytest
Expand Down Expand Up @@ -346,3 +347,314 @@ def test_wrap_method_with_call_not_supported():
def test__deduplicate_metadata_tokens(headers, expected):
dedup = google.api_core.gapic_v1.method._deduplicate_metadata_tokens
assert dedup(*headers) == expected


def test_wrap_method_otel_tracing_disabled(monkeypatch):
"""Proves that when OpenTelemetry tracing is disabled, no span is created."""
monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "false")
mock_target = mock.Mock(return_value="success")
wrapped = google.api_core.gapic_v1.method.wrap_method(mock_target)

with mock.patch(
"google.api_core._observability.is_otel_capabilities_enabled",
return_value=False,
):
assert wrapped() == "success"
mock_target.assert_called_once()


def test_wrap_method_otel_tracing_enabled_success(monkeypatch):
"""Proves that when OpenTelemetry tracing is enabled, a T3 client span is started."""
monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
mock_target = mock.Mock(return_value="success")
mock_target._method = (
"/google.cloud.secretmanager.v1.SecretManagerService/ListSecrets"
)

mock_span = mock.MagicMock()
mock_tracer = mock.MagicMock()
mock_tracer.start_as_current_span.return_value.__enter__.return_value = mock_span

mock_trace = mock.Mock()
mock_trace.get_tracer.return_value = mock_tracer
mock_trace.SpanKind.CLIENT = "CLIENT"

wrapped = google.api_core.gapic_v1.method.wrap_method(mock_target)

with (
mock.patch(
"google.api_core._observability.is_otel_capabilities_enabled",
return_value=True,
),
mock.patch.dict(
sys.modules,
{
"opentelemetry": mock.Mock(trace=mock_trace),
"opentelemetry.trace": mock_trace,
},
),
):
result = wrapped()

assert result == "success"
mock_tracer.start_as_current_span.assert_called_once_with(
"google.cloud.secretmanager.v1.SecretManagerService/ListSecrets",
kind="CLIENT",
attributes={
"rpc.system": "grpc",
"rpc.service": "google.cloud.secretmanager.v1.SecretManagerService",
"rpc.method": "ListSecrets",
},
)


def test_wrap_method_otel_tracing_enabled_error(monkeypatch):
"""Proves that when an RPC fails, the T3 client span records the exception and error status."""
monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
err = RuntimeError("gRPC connection reset")
mock_target = mock.Mock(side_effect=err)
mock_target._method = (
"/google.cloud.secretmanager.v1.SecretManagerService/ListSecrets"
)

mock_span = mock.MagicMock()
mock_tracer = mock.MagicMock()
mock_tracer.start_as_current_span.return_value.__enter__.return_value = mock_span

mock_trace = mock.Mock()
mock_trace.get_tracer.return_value = mock_tracer
mock_trace.SpanKind.CLIENT = "CLIENT"
mock_trace.StatusCode.ERROR = "ERROR"

wrapped = google.api_core.gapic_v1.method.wrap_method(mock_target)

with (
mock.patch(
"google.api_core._observability.is_otel_capabilities_enabled",
return_value=True,
),
mock.patch.dict(
sys.modules,
{
"opentelemetry": mock.Mock(trace=mock_trace),
"opentelemetry.trace": mock_trace,
},
),
):
with pytest.raises(RuntimeError):
wrapped()

mock_span.record_exception.assert_called_once_with(err)
mock_span.set_status.assert_called_once_with("ERROR", str(err))


def test_wrap_method_otel_tracing_bytes_method(monkeypatch):
"""Proves that when raw _method is bytes, it is decoded properly to utf-8."""
monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
mock_target = mock.Mock(return_value="success")
mock_target._method = (
b"/google.cloud.secretmanager.v1.SecretManagerService/ListSecrets"
)

mock_span = mock.MagicMock()
mock_tracer = mock.MagicMock()
mock_tracer.start_as_current_span.return_value.__enter__.return_value = mock_span

mock_trace = mock.Mock()
mock_trace.get_tracer.return_value = mock_tracer
mock_trace.SpanKind.CLIENT = "CLIENT"

wrapped = google.api_core.gapic_v1.method.wrap_method(
mock_target, default_timeout=60
)

with (
mock.patch(
"google.api_core._observability.is_otel_capabilities_enabled",
return_value=True,
),
mock.patch.dict(
sys.modules,
{
"opentelemetry": mock.Mock(trace=mock_trace),
"opentelemetry.trace": mock_trace,
},
),
):
result = wrapped()

assert result == "success"
mock_tracer.start_as_current_span.assert_called_once_with(
"google.cloud.secretmanager.v1.SecretManagerService/ListSecrets",
kind="CLIENT",
attributes={
"rpc.system": "grpc",
"rpc.service": "google.cloud.secretmanager.v1.SecretManagerService",
"rpc.method": "ListSecrets",
},
)


def test_wrap_method_otel_tracing_fallback_with_name(monkeypatch):
"""Proves that when raw _method is absent, fallback uses target.__name__."""
monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")

def custom_rpc(*args, **kwargs):
return "success"

mock_span = mock.MagicMock()
mock_tracer = mock.MagicMock()
mock_tracer.start_as_current_span.return_value.__enter__.return_value = mock_span

mock_trace = mock.Mock()
mock_trace.get_tracer.return_value = mock_tracer
mock_trace.SpanKind.CLIENT = "CLIENT"

wrapped = google.api_core.gapic_v1.method.wrap_method(custom_rpc)

with (
mock.patch(
"google.api_core._observability.is_otel_capabilities_enabled",
return_value=True,
),
mock.patch.dict(
sys.modules,
{
"opentelemetry": mock.Mock(trace=mock_trace),
"opentelemetry.trace": mock_trace,
},
),
):
result = wrapped()

assert result == "success"
mock_tracer.start_as_current_span.assert_called_once_with(
"google.api_core/custom_rpc",
kind="CLIENT",
attributes={
"rpc.system": "grpc",
"rpc.service": "google.api_core",
"rpc.method": "custom_rpc",
},
)


def test_wrap_method_otel_tracing_fallback_without_name(monkeypatch):
"""Proves that when raw _method is absent and target has no explicit __name__,
fallback uses target class name assigned by error wrapper.
"""
monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")

class TargetWithoutName:
def __call__(self, *args, **kwargs):
return "success"

mock_span = mock.MagicMock()
mock_tracer = mock.MagicMock()
mock_tracer.start_as_current_span.return_value.__enter__.return_value = mock_span

mock_trace = mock.Mock()
mock_trace.get_tracer.return_value = mock_tracer
mock_trace.SpanKind.CLIENT = "CLIENT"

wrapped = google.api_core.gapic_v1.method.wrap_method(TargetWithoutName())

with (
mock.patch(
"google.api_core._observability.is_otel_capabilities_enabled",
return_value=True,
),
mock.patch.dict(
sys.modules,
{
"opentelemetry": mock.Mock(trace=mock_trace),
"opentelemetry.trace": mock_trace,
},
),
):
result = wrapped()

assert result == "success"
mock_tracer.start_as_current_span.assert_called_once_with(
"google.api_core/TargetWithoutName",
kind="CLIENT",
attributes={
"rpc.system": "grpc",
"rpc.service": "google.api_core",
"rpc.method": "TargetWithoutName",
},
)


def test_gapic_callable_otel_tracing_fallback_call_default(monkeypatch):
"""Proves that _GapicCallable defaults method to 'call' if target has no __name__."""
monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")

class NoNameTarget:
def __call__(self, *args, **kwargs):
return "success"

target = NoNameTarget()
callable_obj = google.api_core.gapic_v1.method._GapicCallable(
target, None, None, None
)

mock_span = mock.MagicMock()
mock_tracer = mock.MagicMock()
mock_tracer.start_as_current_span.return_value.__enter__.return_value = mock_span

mock_trace = mock.Mock()
mock_trace.get_tracer.return_value = mock_tracer
mock_trace.SpanKind.CLIENT = "CLIENT"

with (
mock.patch(
"google.api_core._observability.is_otel_capabilities_enabled",
return_value=True,
),
mock.patch.dict(
sys.modules,
{
"opentelemetry": mock.Mock(trace=mock_trace),
"opentelemetry.trace": mock_trace,
},
),
):
result = callable_obj()

assert result == "success"
mock_tracer.start_as_current_span.assert_called_once_with(
"google.api_core/call",
kind="CLIENT",
attributes={
"rpc.system": "grpc",
"rpc.service": "google.api_core",
"rpc.method": "call",
},
)


def test_wrap_method_otel_tracing_import_error(monkeypatch):
"""Proves that if opentelemetry raises ImportError, execution proceeds gracefully."""
monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true")
mock_target = mock.Mock(return_value="success")

wrapped = google.api_core.gapic_v1.method.wrap_method(mock_target)

with (
mock.patch(
"google.api_core._observability.is_otel_capabilities_enabled",
return_value=True,
),
mock.patch.dict(
sys.modules,
{
"opentelemetry": None,
"opentelemetry.trace": None,
},
),
):
result = wrapped()

assert result == "success"
mock_target.assert_called_once()
Loading