-
Notifications
You must be signed in to change notification settings - Fork 9
feat(tracing): propagate OTel trace context across Temporal boundaries #485
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.
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
Merged
+125
−1
Merged
Changes from all commits
Commits
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,73 @@ | ||
| """OpenTelemetry trace-context propagation across Temporal boundaries. | ||
|
|
||
| Temporal serializes ``start_workflow`` / ``execute_activity`` across (potentially | ||
| cross-process) boundaries, and does NOT carry the active W3C ``traceparent`` by | ||
| default. So any span created inside a workflow or activity becomes a **new | ||
| detached root** -- the trace shatters at every Temporal hop. | ||
|
|
||
| This bites agentex directly: ``adk.tracing.span`` runs span creation as a | ||
| Temporal activity when ``in_temporal_workflow()`` is true, so without propagation | ||
| those business spans detach from the turn's obs trace. | ||
|
|
||
| Wiring temporalio's first-party ``TracingInterceptor`` onto the Temporal client | ||
| and worker injects the active span context into Temporal headers on the caller | ||
| side and extracts + continues it on the workflow/activity side, using the global | ||
| OpenTelemetry propagator -- so ``client -> workflow -> activity`` is one trace. | ||
|
|
||
| Enabled by DEFAULT. Set ``AGENTEX_TEMPORAL_TRACE_INTERCEPTOR_ENABLED=false`` | ||
| (also accepts ``0`` / ``no`` / ``off``) to turn it off. It also degrades to a | ||
| no-op -- and never raises -- if temporalio's OpenTelemetry contrib isn't | ||
| importable, so enabling it by default can't break a worker. | ||
| """ | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| import os | ||
| from typing import Any | ||
|
|
||
| from agentex.lib.utils.logging import make_logger | ||
|
|
||
| logger = make_logger(__name__) | ||
|
|
||
| _ENABLE_ENV = "AGENTEX_TEMPORAL_TRACE_INTERCEPTOR_ENABLED" | ||
| _FALSEY = {"0", "false", "no", "off"} | ||
|
|
||
|
|
||
| def temporal_trace_interceptor_enabled() -> bool: | ||
| """Whether the Temporal OTel trace interceptor should be installed. | ||
|
|
||
| Defaults to True; disabled only when ``AGENTEX_TEMPORAL_TRACE_INTERCEPTOR_ENABLED`` | ||
| is set to a falsy value (``0`` / ``false`` / ``no`` / ``off``).""" | ||
| return os.environ.get(_ENABLE_ENV, "true").strip().lower() not in _FALSEY | ||
|
|
||
|
|
||
| def temporal_tracing_interceptors() -> list[Any]: | ||
| """Interceptors that propagate OpenTelemetry trace context across Temporal. | ||
|
|
||
| Returns ``[TracingInterceptor()]`` (enabled by default) so callers can splat | ||
| it into a client's / worker's ``interceptors=`` list. Returns ``[]`` when | ||
| disabled via env, or when temporalio's OpenTelemetry contrib is not | ||
| importable. Never raises -- observability wiring must not break a worker. | ||
|
|
||
| ``TracingInterceptor`` implements both the client and worker interceptor | ||
| interfaces, so the same call is used on both sides: | ||
| - on the **client**, it injects context on outbound ``start_workflow`` / | ||
| ``execute_activity`` calls; | ||
| - on the **worker**, it extracts context and roots the workflow / activity | ||
| execution spans under it. | ||
| """ | ||
| if not temporal_trace_interceptor_enabled(): | ||
| logger.info("Temporal OTel trace interceptor disabled via %s", _ENABLE_ENV) | ||
| return [] | ||
| try: | ||
| from temporalio.contrib.opentelemetry import TracingInterceptor | ||
|
|
||
| # Construct inside the try so a constructor failure (not just a missing | ||
| # contrib) also falls back to a no-op instead of aborting worker startup. | ||
| return [TracingInterceptor()] | ||
| except Exception as exc: # contrib unavailable OR constructor failure -> no-op, never raise | ||
| logger.warning( | ||
| "Temporal OTel trace interceptor unavailable (%s); traces will not propagate across Temporal boundaries.", | ||
| exc, | ||
| ) | ||
| return [] |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,40 @@ | ||
| """Unit tests for the Temporal OTel trace-interceptor wiring. | ||
|
|
||
| Verifies the interceptor is on by default, the opt-out env flag, and the safe | ||
| no-op fallback when temporalio's OpenTelemetry contrib isn't importable. | ||
| """ | ||
|
|
||
| import sys | ||
|
|
||
| import pytest | ||
|
|
||
| from agentex.lib.core.tracing import temporal as temporal_tracing | ||
|
|
||
|
|
||
| class TestTemporalTraceInterceptor: | ||
| def test_enabled_by_default(self, monkeypatch): | ||
| monkeypatch.delenv("AGENTEX_TEMPORAL_TRACE_INTERCEPTOR_ENABLED", raising=False) | ||
| assert temporal_tracing.temporal_trace_interceptor_enabled() is True | ||
|
|
||
| interceptors = temporal_tracing.temporal_tracing_interceptors() | ||
| assert len(interceptors) == 1 | ||
| # temporalio's first-party OTel interceptor | ||
| assert type(interceptors[0]).__name__ == "TracingInterceptor" | ||
|
|
||
| @pytest.mark.parametrize("value", ["false", "0", "no", "off", "FALSE", "Off"]) | ||
| def test_disabled_via_env(self, monkeypatch, value): | ||
| monkeypatch.setenv("AGENTEX_TEMPORAL_TRACE_INTERCEPTOR_ENABLED", value) | ||
| assert temporal_tracing.temporal_trace_interceptor_enabled() is False | ||
| assert temporal_tracing.temporal_tracing_interceptors() == [] | ||
|
|
||
| @pytest.mark.parametrize("value", ["true", "1", "yes", "TRUE", "anything"]) | ||
| def test_enabled_for_non_falsy_values(self, monkeypatch, value): | ||
| monkeypatch.setenv("AGENTEX_TEMPORAL_TRACE_INTERCEPTOR_ENABLED", value) | ||
| assert temporal_tracing.temporal_trace_interceptor_enabled() is True | ||
|
|
||
| def test_no_op_when_contrib_unimportable(self, monkeypatch): | ||
| # Enabled, but temporalio's OTel contrib not importable -> [] (never raises), | ||
| # so default-on can't break a worker that lacks the contrib. | ||
| monkeypatch.delenv("AGENTEX_TEMPORAL_TRACE_INTERCEPTOR_ENABLED", raising=False) | ||
| monkeypatch.setitem(sys.modules, "temporalio.contrib.opentelemetry", None) | ||
| assert temporal_tracing.temporal_tracing_interceptors() == [] |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This double installs the tracing interceptor on the worker. Worker.init prepends client interceptors that also implement temporalio.worker.Interceptor, and TracingInterceptor implements both, so the instance already on this worker's client (from get_temporal_client above) applies to the worker automatically. With this line the worker runs two instances. Verified with a local repro on temporalio 1.26.0: every worker side span (RunWorkflow, CompleteWorkflow, StartActivity, RunActivity) is emitted twice, while client only wiring emits each span once with propagation intact.
Suggest interceptors=self.interceptors. The inherited instance is prepended, so tracing still sits outermost ahead of the business interceptors.