Keep spans mutable during on_ending span processor callbacks

.changelog/5613.fixed

@@ -0,0 +1 @@
+`opentelemetry-sdk`: keep spans mutable during `on_ending` span processor callbacks. Only the thread that called `end()` may mutate the span while the callbacks run, so `ConcurrentMultiSpanProcessor` now invokes `_on_ending` synchronously on that thread instead of through its thread pool.

opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py

@@ -226,7 +226,9 @@ class ConcurrentMultiSpanProcessor(SpanProcessor):
 
     Calls to the underlying span processors are forwarded in parallel by
     submitting them to a thread pool executor and waiting until each span
-    processor finished its work.
+    processor finished its work. The only exception is ``_on_ending``, which
+    is called synchronously on the thread ending the span, since that is the
+    only thread allowed to mutate the span while it is ending.
 
     Args:
         num_threads: The number of threads managed by the thread pool executor
@@ -282,8 +284,12 @@ def on_start(
         self._submit_and_await(lambda sp: sp.on_start, span, parent_context=parent_context)
 
     def _on_ending(self, span: "Span") -> None:
-        # pylint: disable=protected-access
-        self._submit_and_await(lambda sp: sp._on_ending, span)
+        # Unlike the other callbacks, _on_ending runs synchronously on the
+        # thread that ends the span: that is the only thread allowed to
+        # mutate the span while it is ending.
+        for sp in self._span_processors:
+            # pylint: disable=protected-access
+            sp._on_ending(span)
 
     def on_end(self, span: "ReadableSpan") -> None:
         self._submit_and_await(lambda sp: sp.on_end, span)
@@ -378,7 +384,7 @@ def _check_span_ended(func):
     def wrapper(self, *args, **kwargs):
         already_ended = False
         with self._lock:  # pylint: disable=protected-access
-            if self._end_time is None:  # pylint: disable=protected-access
+            if self._is_mutable():  # pylint: disable=protected-access
                 func(self, *args, **kwargs)
             else:
                 already_ended = True
@@ -807,6 +813,10 @@ def __init__(
         self._span_processor = span_processor
         self._limits = limits
         self._lock = threading.Lock()
+        # Identifier of the thread running end() while the _on_ending
+        # callbacks execute: the end timestamp is already set, but the span
+        # must remain mutable from that thread until the callbacks return.
+        self._ending_thread: int | None = None
         self._attributes = BoundedAttributes(
             self._limits.max_span_attributes,
             attributes,
@@ -859,15 +869,15 @@ def get_span_context(self) -> trace_api.SpanContext:
 
     def set_attributes(self, attributes: Mapping[str, types.AnyValue]) -> None:
         with self._lock:
-            if self._end_time is not None:
+            if not self._is_mutable():
                 logger.warning("Setting attribute on ended span.")
                 return
 
             self._attributes._set_items(attributes)  # pylint: disable=protected-access
 
     def set_attribute(self, key: str, value: types.AnyValue) -> None:
         with self._lock:
-            if self._end_time is not None:
+            if not self._is_mutable():
                 logger.warning("Setting attribute on ended span.")
                 return
 
@@ -961,21 +971,35 @@ def end(self, end_time: int | None = None) -> None:
                 logger.warning("Calling end() on an ended span.")
                 return
 
+            self._ending_thread = threading.get_ident()
             self._end_time = end_time if end_time is not None else time_ns()
-            self._attributes._immutable = True  # pylint: disable=protected-access
 
-        if self._record_end_metrics:
-            self._record_end_metrics()
+        # The span must remain mutable from this thread while the _on_ending
+        # callbacks run; it becomes immutable once they have returned.
         # pylint: disable=protected-access
-        self._span_processor._on_ending(self)
+        try:
+            if self._record_end_metrics:
+                self._record_end_metrics()
+            self._span_processor._on_ending(self)
+        finally:
+            with self._lock:
+                self._ending_thread = None
+                self._attributes._immutable = True
         self._span_processor.on_end(self._readable_span())
 
     @_check_span_ended
     def update_name(self, name: str) -> None:
         self._name = name
 
+    def _is_mutable(self) -> bool:
+        return self._end_time is None or self._ending_thread == threading.get_ident()
+
     def is_recording(self) -> bool:
-        return self._end_time is None
+        # Unlike the mutators, this check is deliberately not taken under
+        # self._lock: it does not modify the span, and the answer can become
+        # stale as soon as it is returned anyway, since another thread may
+        # end the span right after this call.
+        return self._is_mutable()
 
     @_check_span_ended
     def set_status(

opentelemetry-sdk/tests/trace/test_span_processor.py

@@ -8,7 +8,7 @@
 import time
 import unittest
 import weakref
-from threading import Event
+from threading import Event, Thread
 from unittest import mock
 
 from opentelemetry import trace as trace_api
@@ -251,6 +251,157 @@ def test_on_ending_not_implemented_does_not_raise(self):
 
         self.assertListEqual(spans_calls_list, expected_list)
 
+    def test_span_mutable_during_on_ending(self):
+        exporter = InMemorySpanExporter()
+
+        class MutatingSpanProcessor(trace.SpanProcessor):
+            def _on_ending(self, span: "trace.Span") -> None:
+                assert span.is_recording()
+                assert span.end_time is not None
+                span.update_name("renamed")
+                span.set_attribute("attribute", "value")
+                span.set_attributes({"attributes": "value"})
+                span.add_event("event")
+                span.add_link(
+                    trace_api.SpanContext(
+                        trace_id=0x1,
+                        span_id=0x2,
+                        is_remote=False,
+                    )
+                )
+                span.set_status(trace_api.StatusCode.ERROR)
+
+        tracer_provider = trace.TracerProvider()
+        tracer_provider.add_span_processor(MutatingSpanProcessor())
+        tracer_provider.add_span_processor(SimpleSpanProcessor(exporter))
+        tracer = tracer_provider.get_tracer(__name__)
+
+        tracer.start_span("foo").end()
+
+        (span,) = exporter.get_finished_spans()
+        self.assertEqual(span.name, "renamed")
+        self.assertEqual(span.attributes["attribute"], "value")
+        self.assertEqual(span.attributes["attributes"], "value")
+        self.assertEqual(span.events[0].name, "event")
+        self.assertEqual(len(span.links), 1)
+        self.assertIs(span.status.status_code, trace_api.StatusCode.ERROR)
+
+    def test_end_during_on_ending_is_ignored(self):
+        exporter = InMemorySpanExporter()
+        on_ending_calls = []
+
+        class ReentrantSpanProcessor(trace.SpanProcessor):
+            def _on_ending(self, span: "trace.Span") -> None:
+                on_ending_calls.append(span)
+                span.end()
+
+        tracer_provider = trace.TracerProvider()
+        tracer_provider.add_span_processor(ReentrantSpanProcessor())
+        tracer_provider.add_span_processor(SimpleSpanProcessor(exporter))
+        tracer = tracer_provider.get_tracer(__name__)
+
+        span = tracer.start_span("foo")
+        span.end()
+
+        self.assertEqual(len(on_ending_calls), 1)
+        self.assertEqual(len(exporter.get_finished_spans()), 1)
+
+        # After end() returns the span is immutable again.
+        self.assertFalse(span.is_recording())
+        span.set_attribute("late", "value")
+        (exported,) = exporter.get_finished_spans()
+        self.assertNotIn("late", exported.attributes)
+
+    def test_other_thread_cannot_mutate_during_on_ending(self):
+        exporter = InMemorySpanExporter()
+        inside_on_ending = Event()
+        other_thread_done = Event()
+
+        class BlockingSpanProcessor(trace.SpanProcessor):
+            def _on_ending(self, span: "trace.Span") -> None:
+                inside_on_ending.set()
+                other_thread_done.wait(5)
+
+        tracer_provider = trace.TracerProvider()
+        tracer_provider.add_span_processor(BlockingSpanProcessor())
+        tracer_provider.add_span_processor(SimpleSpanProcessor(exporter))
+        span = tracer_provider.get_tracer(__name__).start_span("foo")
+
+        def mutate_from_another_thread():
+            inside_on_ending.wait(5)
+            span.set_attribute("other.attribute", "value")
+            span.set_attributes({"other.attributes": "value"})
+            span.update_name("renamed")
+            span.add_event("event")
+            span.add_link(
+                trace_api.SpanContext(
+                    trace_id=0x1,
+                    span_id=0x2,
+                    is_remote=False,
+                )
+            )
+            span.set_status(trace_api.StatusCode.ERROR)
+            other_thread_done.set()
+
+        thread = Thread(target=mutate_from_another_thread)
+        thread.start()
+        span.end()
+        thread.join(5)
+
+        self.assertTrue(inside_on_ending.is_set())
+        self.assertTrue(other_thread_done.is_set())
+
+        (exported,) = exporter.get_finished_spans()
+        self.assertNotIn("other.attribute", exported.attributes)
+        self.assertNotIn("other.attributes", exported.attributes)
+        self.assertEqual(exported.name, "foo")
+        self.assertEqual(len(exported.events), 0)
+        self.assertEqual(len(exported.links), 0)
+        self.assertIs(exported.status.status_code, trace_api.StatusCode.UNSET)
+
+    def test_other_thread_does_not_see_span_recording_during_on_ending(self):
+        inside_on_ending = Event()
+        other_thread_done = Event()
+        recording_seen_by_other_thread = []
+
+        class BlockingSpanProcessor(trace.SpanProcessor):
+            def _on_ending(self, span: "trace.Span") -> None:
+                inside_on_ending.set()
+                other_thread_done.wait(5)
+
+        tracer_provider = trace.TracerProvider()
+        tracer_provider.add_span_processor(BlockingSpanProcessor())
+        span = tracer_provider.get_tracer(__name__).start_span("foo")
+
+        def observe_from_another_thread():
+            inside_on_ending.wait(5)
+            recording_seen_by_other_thread.append(span.is_recording())
+            other_thread_done.set()
+
+        thread = Thread(target=observe_from_another_thread)
+        thread.start()
+        span.end()
+        thread.join(5)
+
+        self.assertTrue(other_thread_done.is_set())
+        self.assertEqual(recording_seen_by_other_thread, [False])
+
+    def test_span_is_frozen_when_record_end_metrics_raises(self):
+        def record_end_metrics_that_raises():
+            raise RuntimeError("meter blew up")
+
+        tracer_provider = trace.TracerProvider()
+        span = tracer_provider.get_tracer(__name__).start_span("foo")
+        # pylint: disable=protected-access
+        span._record_end_metrics = record_end_metrics_that_raises
+
+        with self.assertRaises(RuntimeError):
+            span.end()
+
+        self.assertFalse(span.is_recording())
+        span.set_attribute("late", "value")
+        self.assertNotIn("late", span.attributes)
+
 
 class MultiSpanProcessorTestBase(abc.ABC):
     @abc.abstractmethod
@@ -371,6 +522,25 @@ def test_on_ending_not_implemented_does_not_raise(self):
         # pylint: disable=no-member
         self.assertListEqual(spans_calls_list, expected_list)
 
+    def test_span_mutable_during_on_ending(self):
+        multi_processor = self.create_multi_span_processor()
+        exporter = InMemorySpanExporter()
+
+        class MutatingSpanProcessor(trace.SpanProcessor):
+            def _on_ending(self, span: "trace.Span") -> None:
+                span.set_attribute("attribute", "value")
+
+        multi_processor.add_span_processor(MutatingSpanProcessor())
+        multi_processor.add_span_processor(SimpleSpanProcessor(exporter))
+        tracer_provider = trace.TracerProvider(active_span_processor=multi_processor)
+
+        tracer_provider.get_tracer(__name__).start_span("foo").end()
+
+        (span,) = exporter.get_finished_spans()
+        # pylint: disable=no-member
+        self.assertEqual(span.attributes["attribute"], "value")
+        multi_processor.shutdown()
+
 
 class TestSynchronousMultiSpanProcessor(MultiSpanProcessorTestBase, unittest.TestCase):
     def create_multi_span_processor(