feat(api-core): add eager channel orchestration for OpenTelemetry - #18237
feat(api-core): add eager channel orchestration for OpenTelemetry#18237chalmerlowe wants to merge 2 commits into
Conversation
- Implement create_channel_with_otel and create_async_channel_with_otel helpers - Deduplicate interceptor instantiation via internal _get_otel_interceptor - Add unit tests in test_observability.py
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request refactors OpenTelemetry channel instrumentation in google/api_core/_observability.py by replacing apply_otel_capabilities_to_channel with dedicated helpers for creating synchronous and asynchronous channels with OTel capabilities (create_channel_with_otel and create_async_channel_with_otel). Unit tests are updated accordingly. The review feedback highlights an inconsistency in interceptor execution order between the sync and async implementations, suggesting that the async OTel interceptor should be prepended rather than appended to the interceptors list to maintain consistent tracing semantics across both environments.
| interceptors = list(channel_kwargs.pop("interceptors", []) or []) | ||
| interceptors.append(async_interceptor) | ||
| channel_kwargs["interceptors"] = interceptors |
There was a problem hiding this comment.
There is an inconsistency in the interceptor execution order between the synchronous and asynchronous implementations:
- In
create_channel_with_otel(sync), the raw channel is first created (with any user interceptors already applied by the factory), and then wrapped with the OpenTelemetry interceptor. This makes the OpenTelemetry interceptor the outermost wrapper (it executes first on outbound requests). - In
create_async_channel_with_otel(async), the OpenTelemetry interceptor is appended to the end of theinterceptorslist, making it the innermost wrapper (it executes last, closest to the network call).
To ensure consistent behavior and tracing semantics (e.g., capturing latency/errors introduced by other user interceptors) across both sync and async channels, we should prepend the OpenTelemetry interceptor to the list of interceptors in the async case.
| interceptors = list(channel_kwargs.pop("interceptors", []) or []) | |
| interceptors.append(async_interceptor) | |
| channel_kwargs["interceptors"] = interceptors | |
| interceptors = list(channel_kwargs.pop("interceptors", []) or []) | |
| interceptors.insert(0, async_interceptor) | |
| channel_kwargs["interceptors"] = interceptors |
There was a problem hiding this comment.
did you consider the external constraints on this?
namely:
In synchronous calls, we have to create the channel first using the otel-grpc constructor and we then pass the channel on to grpc for further wrapping with all the other interceptors. This happens because in synchronous calls otel-grpc produces an interceptor that grpc does not have the ability to apply (it outputs a TypeError). In the sync environment, channels are mutable, so this works. Because our otel wrapper gets applied when we create the channel it is inherently the inner-most wrapper.
In async calls the type mismatch is not a problem BUT the channels are immutable, so we have to send all the interceptors to grpc as a batch. Because our otel wrapper is last and grpc reverses the list, when the channel gets created it is again, inherently the inner-most wrapper.
This is why the two functions are asymmetric: they need to handle things differently because the grpc library accepts inputs differently.
There was a problem hiding this comment.
Thank you for the clarification, @chalmerlowe. You are absolutely correct regarding the fundamental differences in how grpc and grpc.aio handle interceptors and channel construction.
Given that grpc.aio channels are immutable and the interceptors are applied in the order they appear in the list (with the last one being the innermost), your implementation of appending the OTel interceptor is indeed the correct way to ensure it acts as the innermost wrapper. My previous suggestion to prepend it would have incorrectly placed it as the outermost wrapper in the async context.
I appreciate you highlighting these constraints.
| mock_channel = mock.Mock() | ||
| mock_intercepted_channel = mock.Mock() | ||
| def test_get_otel_interceptor_sync_default(monkeypatch): | ||
| mock_otel = mock.Mock() |
There was a problem hiding this comment.
There is plenty of room for deduplicating some of the inner workings of these tests (using fixtures, reusable functions, etc). Happy to revise these but would prefer to get some initial buy-in on the overall approach in the body of the code before investing in what might end up being premature optimization.
… tests
- Use list(channel_kwargs.pop('interceptors', None) or []) in create_async_channel_with_otel
- Add unit tests for None and omitted interceptors arguments
| def create_channel_with_otel( | ||
| channel_factory: Callable[..., Any], | ||
| client_options: Optional[Union[ClientOptions, dict[str, Any]]] = None, | ||
| **channel_kwargs: Any, |
There was a problem hiding this comment.
Can we also accept *channel_args? That would make this easier to pass into the transport init: #18188 (comment)
| return otel_grpc.client_interceptor(tracer_provider=tracer_provider) | ||
|
|
||
|
|
||
| def create_channel_with_otel( |
There was a problem hiding this comment.
I left some comments in your other PR, but if it's possible to decouple the interceptor more from the channel, that could make thinks a lot easier for composition in the future.
I think the previous apply_otel_capabilities_to_channel would be better suited for this. If we go with option A, the client could do something like
grpc_interceptor = functools.partial(apply_otel_capabilities_to_channel, client_options=options)
interceptor_list = [grpc_interceptor, logging_interceptor]
Transport(interceptors=interceptor_list, ...)
2 participants
This pull request introduces channel orchestration helper functions in
google.api_core._observabilityto support OpenTelemetry (OTel) client interceptors for synchronous gRPC channels (asynchronous gRPC channels are included for comparison).Problem
Generated client libraries need a consistent and centralized way to create and instrument gRPC channels with OpenTelemetry tracing when enabled via environment variables or client options. In addition, synchronous and asynchronous gRPC handle interceptors differently:
grpc) allows channels to be intercepted after they've been constructed.grpc.aio) requires interceptors to be passed during channel construction.Solution
Added
create_channel_with_otel:channel_factorywith keyword arguments.Added
create_async_channel_with_otel:interceptorsparameter before calling thechannel_factory.Added internal
_get_otel_interceptor:tracer_providerfromClientOptionsor configuration dictionaries.client_interceptor) or asynchronous (aio_client_interceptor) interceptors without code duplication.Added unit test suite in
tests/unit/test_observability.py:Notes for Reviewers
google-api-corewithout altering existing client behavior until clients opt into calling these helpers.create_channel_with_otelandcreate_async_channel_with_otelreflects the underlying immutability requirements ofgrpc.aiochannels compared to synchronousgrpcchannels and is provided solely for comparison during review.