Skip to content

Commit df6f4ed

Browse files
authored
Stop spurious Failed to detach context errors from the OpenTelemetry interceptors (#1911)
* Detach OpenTelemetry context only on the thread that attached it The tracing interceptor's best-effort detach checked that the attached Context was still current, so a generator finalized by GC on another thread would skip the detach instead of using a token that is invalid there. OpenTelemetry's threading instrumentation, which strands enables whenever an Agent is created, copies the same Context object into new threads, so that check passes on the wrong thread and detach logs "Failed to detach context". Record the attaching thread with the token and skip the detach anywhere else. The regression test runs the existing safe-detach scenario with the instrumentation active; it fails without this change. * Cover OpenTelemetryInterceptor and compare the attaching Thread object OpenTelemetryPlugin installs OpenTelemetryInterceptor, whose four attach/detach sites had the same context-identity guard. Move the helpers into _context.py and use them from both interceptors. Compare the attaching threading.Thread object rather than its id, which the OS can reuse. The regression test is now a matrix over interceptor and threading instrumentation; the OpenTelemetryInterceptor case fails without this. * Detach where the token is valid, not where the thread matches Workflow activations run on a thread pool while the asyncio task keeps its contextvars.Context, so a context attached in one activation is legitimately detached in a later one on another thread. The thread check skipped those detaches and left the inner context attached for outer interceptors. Let contextvars decide instead: perform the reset that opentelemetry.context.detach performs and treat its ValueError for a foreign Context as nothing to detach. Covers the thread change with a unit test on both interceptors and a worker test whose executor alternates activations between two threads. * Patch AttachedContext.detach through monkeypatch in the leak test mypy rejects assigning to a method (method-assign); pytest's monkeypatch does the same thing and restores it on teardown. * Make the attach/detach helper a context manager attached_context(context) replaces the seven attach/try/finally/detach sites; the guarded detach moves to _detach, which the leak test counts. * Count the interceptors' own detaches in the activity leak test The test patched opentelemetry.context.attach/detach process-wide, which also counts the threading instrumentation's attach/detach around every thread's run(). A pool thread started before the patch and exiting during the test adds a detach with no counted attach (9 attaches vs 10 detaches on 3.10 macos-arm). Record the interceptors' own detach results instead.
1 parent cf11a1d commit df6f4ed

5 files changed

Lines changed: 316 additions & 110 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,11 @@ to include examples, links to docs, or any other relevant information.
3636

3737
### Fixed
3838

39+
- `temporalio.contrib.opentelemetry`: `TracingInterceptor` and `OpenTelemetryInterceptor` no longer
40+
log `Failed to detach context` when a context is torn down on a different thread while
41+
OpenTelemetry's threading instrumentation (enabled by strands, among others) is active; a
42+
context is now detached exactly when its token is still valid in the current
43+
`contextvars.Context`, which it stays when a workflow resumes on another pool thread.
3944
### Security
4045

4146
## [1.34.0] - 2026-09-30
Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
"""Attach OpenTelemetry contexts so that the matching detach is always safe."""
2+
3+
from __future__ import annotations
4+
5+
from collections.abc import Iterator
6+
from contextlib import contextmanager
7+
from contextvars import Token
8+
9+
import opentelemetry.context
10+
from opentelemetry.context import Context
11+
12+
13+
@contextmanager
14+
def attached_context(context: Context | None) -> Iterator[None]:
15+
"""Attach ``context`` for the block and detach it afterwards where possible.
16+
17+
``None`` attaches nothing. The block's ``finally`` can run in a different
18+
``contextvars.Context`` than the one that attached: a context manager
19+
abandoned by an evicted workflow is finalized wherever garbage collection
20+
happens to run. The token is not valid there, and
21+
``opentelemetry.context.detach`` would log "Failed to detach context" even
22+
though there is nothing to detach. Checking that the attached context is
23+
still current does not catch every such case, because OpenTelemetry's
24+
threading instrumentation (enabled by strands, among others) propagates
25+
the same ``Context`` object into new threads. The thread is no test
26+
either: workflow activations move between pool threads while the asyncio
27+
task keeps its ``contextvars.Context``, and those detaches must happen.
28+
Only ``contextvars`` knows which ``Context`` a token belongs to, so
29+
:func:`_detach` performs the reset ``detach`` performs and ignores the
30+
``ValueError`` raised for a token from another ``Context``.
31+
"""
32+
if context is None:
33+
yield
34+
return
35+
token = opentelemetry.context.attach(context)
36+
try:
37+
yield
38+
finally:
39+
_detach(context, token)
40+
41+
42+
def _detach(context: Context, token: Token[Context]) -> bool:
43+
"""Detach ``context`` if it is current and ``token`` is valid here."""
44+
if context is not opentelemetry.context.get_current():
45+
return False
46+
try:
47+
token.var.reset(token)
48+
except ValueError:
49+
# The token was created in a different contextvars.Context.
50+
return False
51+
return True

‎temporalio/contrib/opentelemetry/_interceptor.py‎

Lines changed: 20 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
import temporalio.nexus.system.workflow_service.models
3737
import temporalio.worker
3838
import temporalio.workflow
39+
from temporalio.contrib.opentelemetry._context import attached_context
3940
from temporalio.exceptions import ApplicationError, ApplicationErrorCategory
4041

4142
# OpenTelemetry dynamically, lazily chooses its context implementation at
@@ -183,8 +184,7 @@ def _start_as_current_span(
183184
kind: opentelemetry.trace.SpanKind,
184185
context: Context | None = None,
185186
) -> Iterator[None]:
186-
token = opentelemetry.context.attach(context) if context else None
187-
try:
187+
with attached_context(context):
188188
with self.tracer.start_as_current_span(
189189
name,
190190
attributes=attributes,
@@ -219,9 +219,6 @@ def _start_as_current_span(
219219
)
220220
)
221221
raise
222-
finally:
223-
if token and context is opentelemetry.context.get_current():
224-
opentelemetry.context.detach(token)
225222

226223
def _completed_workflow_span(
227224
self, params: _CompletedWorkflowSpanParams
@@ -556,8 +553,7 @@ async def handle_query(self, input: temporalio.worker.HandleQueryInput) -> Any:
556553
# We need to put this interceptor on the context too
557554
context = self._set_on_context(context)
558555
# Run under context with new span
559-
token = opentelemetry.context.attach(context)
560-
try:
556+
with attached_context(context):
561557
# This won't be created if there was no context header
562558
self._completed_span(
563559
f"HandleQuery:{input.query}",
@@ -567,13 +563,6 @@ async def handle_query(self, input: temporalio.worker.HandleQueryInput) -> Any:
567563
kind=opentelemetry.trace.SpanKind.SERVER,
568564
)
569565
return await super().handle_query(input)
570-
finally:
571-
# In some exceptional cases this finally is executed with a
572-
# different contextvars.Context than the one the token was created
573-
# on. As such we do a best effort detach to avoid using a mismatched
574-
# token.
575-
if context is opentelemetry.context.get_current():
576-
opentelemetry.context.detach(token)
577566

578567
def handle_update_validator(
579568
self, input: temporalio.worker.HandleUpdateInput
@@ -644,31 +633,23 @@ def _top_level_workflow_context(
644633
success = False
645634
exception: Exception | None = None
646635
# Run under this context
647-
token = opentelemetry.context.attach(context)
648-
649-
try:
650-
yield None
651-
success = True
652-
except temporalio.exceptions.FailureError as err:
653-
# We only record the failure errors since those are the only ones
654-
# that lead to workflow completions
655-
exception = err
656-
raise
657-
finally:
658-
# Create a completed span before detaching context
659-
if exception or (success and success_is_complete):
660-
self._completed_span(
661-
f"CompleteWorkflow:{temporalio.workflow.info().workflow_type}",
662-
exception=exception,
663-
kind=opentelemetry.trace.SpanKind.INTERNAL,
664-
)
665-
666-
# In some exceptional cases this finally is executed with a
667-
# different contextvars.Context than the one the token was created
668-
# on. As such we do a best effort detach to avoid using a mismatched
669-
# token.
670-
if context is opentelemetry.context.get_current():
671-
opentelemetry.context.detach(token)
636+
with attached_context(context):
637+
try:
638+
yield None
639+
success = True
640+
except temporalio.exceptions.FailureError as err:
641+
# We only record the failure errors since those are the only ones
642+
# that lead to workflow completions
643+
exception = err
644+
raise
645+
finally:
646+
# Create a completed span before detaching context
647+
if exception or (success and success_is_complete):
648+
self._completed_span(
649+
f"CompleteWorkflow:{temporalio.workflow.info().workflow_type}",
650+
exception=exception,
651+
kind=opentelemetry.trace.SpanKind.INTERNAL,
652+
)
672653

673654
def _context_to_headers(
674655
self, headers: Mapping[str, temporalio.api.common.v1.Payload]

‎temporalio/contrib/opentelemetry/_otel_interceptor.py‎

Lines changed: 5 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@
3535
import temporalio.nexus.system.workflow_service.models
3636
import temporalio.worker
3737
import temporalio.workflow
38+
from temporalio.contrib.opentelemetry._context import attached_context
3839
from temporalio.contrib.opentelemetry._tracer_provider import (
3940
ReplaySafeTracerProvider,
4041
)
@@ -125,8 +126,7 @@ def _maybe_span(
125126
yield
126127
return
127128

128-
token = opentelemetry.context.attach(context) if context else None
129-
try:
129+
with attached_context(context):
130130
with tracer.start_as_current_span(
131131
name,
132132
attributes=attributes,
@@ -148,9 +148,6 @@ def _maybe_span(
148148
)
149149
)
150150
raise
151-
finally:
152-
if token and context is opentelemetry.context.get_current():
153-
opentelemetry.context.detach(token)
154151

155152

156153
class OpenTelemetryInterceptor(
@@ -334,8 +331,7 @@ async def execute_activity(
334331
self, input: temporalio.worker.ExecuteActivityInput
335332
) -> Any:
336333
context = _headers_to_context(input.headers)
337-
token = opentelemetry.context.attach(context)
338-
try:
334+
with attached_context(context):
339335
info = temporalio.activity.info()
340336
with _maybe_span(
341337
get_tracer(__name__),
@@ -349,9 +345,6 @@ async def execute_activity(
349345
kind=opentelemetry.trace.SpanKind.SERVER,
350346
):
351347
return await super().execute_activity(input)
352-
finally:
353-
if context is opentelemetry.context.get_current():
354-
opentelemetry.context.detach(token)
355348

356349

357350
class _TracingNexusOperationInboundInterceptor(
@@ -368,12 +361,8 @@ def __init__(
368361
@contextmanager
369362
def _top_level_context(self, headers: Mapping[str, str]) -> Iterator[None]:
370363
context = _nexus_headers_to_context(headers)
371-
token = opentelemetry.context.attach(context)
372-
try:
364+
with attached_context(context):
373365
yield
374-
finally:
375-
if context is opentelemetry.context.get_current():
376-
opentelemetry.context.detach(token)
377366

378367
async def execute_nexus_operation_start(
379368
self, input: temporalio.worker.ExecuteNexusOperationStartInput
@@ -501,12 +490,8 @@ async def handle_update_handler(
501490
@contextmanager
502491
def _top_level_workflow_context(self, input: _InputWithHeaders) -> Iterator[None]:
503492
context = _headers_to_context(input.headers)
504-
token = opentelemetry.context.attach(context)
505-
try:
493+
with attached_context(context):
506494
yield
507-
finally:
508-
if context is opentelemetry.context.get_current():
509-
opentelemetry.context.detach(token)
510495

511496

512497
class _TracingWorkflowOutboundInterceptor(

0 commit comments

Comments
 (0)