| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
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
| 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 [] |
| 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() == [] |
| Back | FazBrowse Home | New Git URL |
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.
Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low QualityThis 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.
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.