oss_12_onending_frozen

.changelog/14.fixed

@@ -0,0 +1,3 @@
+Run `SpanProcessor.on_ending` while the span is still mutable so
+attributes, events and links added in the hook are no longer dropped; the
+span is finalized (frozen) only after `on_ending` completes

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

@@ -289,8 +289,16 @@ def on_start(
         )
 
     def _on_ending(self, span: "Span") -> None:
-        # pylint: disable=protected-access
-        self._submit_and_await(lambda sp: sp._on_ending, span)
+        # ``on_ending`` runs while the span is still mutable, but only from the
+        # thread that is ending it. Fanning the hooks out to the thread pool
+        # would let them mutate the span from worker threads, which is both
+        # racy and rejected by the span's ending-thread guard. The hooks are
+        # therefore invoked sequentially on the ending thread. This matches the
+        # documented contract that ``on_ending`` is called synchronously on the
+        # thread that ends the span.
+        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)
@@ -387,7 +395,8 @@ 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
+            # pylint: disable=protected-access
+            if self._is_writable_by_current_thread():
                 func(self, *args, **kwargs)
             else:
                 already_ended = True
@@ -840,6 +849,13 @@ def __init__(
         self._span_processor = span_processor
         self._limits = limits
         self._lock = threading.Lock()
+        # ``_ending`` marks the intermediate state (see the ``end()`` method):
+        # the span has an ``_end_time`` but is not yet finalized. While in this
+        # state the ``on_ending`` hooks run and the ending thread is still
+        # allowed to mutate the span. ``_ending_thread_id`` records which thread
+        # is permitted to do so.
+        self._ending = False
+        self._ending_thread_id: int | None = None
         self._attributes = BoundedAttributes(
             self._limits.max_span_attributes,
             attributes,
@@ -886,19 +902,31 @@ def _new_links(self, links: Sequence[trace_api.Link]):
     def get_span_context(self) -> trace_api.SpanContext:
         return typing.cast(trace_api.SpanContext, self._context)
 
+    def _is_writable_by_current_thread(self) -> bool:
+        """Whether the span may currently be mutated.
+
+        Must be called while holding ``self._lock``. A span is writable while it
+        has not been ended at all, and also during the intermediate "ending"
+        state (while ``on_ending`` hooks run) but only from the thread that is
+        ending the span.
+        """
+        if self._end_time is None:
+            return True
+        return self._ending and self._ending_thread_id == threading.get_ident()
+
     def set_attributes(
         self, attributes: Mapping[str, types.AttributeValue]
     ) -> None:
         with self._lock:
-            if self._end_time is not None:
+            if not self._is_writable_by_current_thread():
                 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.AttributeValue) -> None:
         with self._lock:
-            if self._end_time is not None:
+            if not self._is_writable_by_current_thread():
                 logger.warning("Setting attribute on ended span.")
                 return
 
@@ -991,13 +1019,28 @@ def end(self, end_time: int | None = None) -> None:
                 logger.warning("Calling end() on an ended span.")
                 return
 
+            # Enter the intermediate "ending" state: the span has an end time
+            # but is not frozen yet. The ``on_ending`` hooks run in this state
+            # and this thread is still allowed to mutate the span, so its
+            # attributes are intentionally left mutable for now.
             self._end_time = end_time if end_time is not None else time_ns()
-            self._attributes._immutable = True  # pylint: disable=protected-access
+            self._ending = True
+            self._ending_thread_id = threading.get_ident()
 
         if self._record_end_metrics:
             self._record_end_metrics()
-        # pylint: disable=protected-access
-        self._span_processor._on_ending(self)
+
+        try:
+            # pylint: disable=protected-access
+            self._span_processor._on_ending(self)
+        finally:
+            # Finalize (freeze) the span. After this point mutations are
+            # rejected even from the ending thread.
+            with self._lock:
+                self._ending = False
+                self._ending_thread_id = None
+                self._attributes._immutable = True  # pylint: disable=protected-access
+
         self._span_processor.on_end(self._readable_span())
 
     @_check_span_ended

opentelemetry-sdk/tests/trace/test_span_processor.py

@@ -51,6 +51,23 @@ def _on_ending(self, span: "trace.Span") -> None:
         self.span_list.append(span_event_ending_fmt(self.name, span.name))
 
 
+class MutatingOnEndingSpanProcessor(trace.SpanProcessor):
+    """Span processor that mutates the span from inside ``on_ending``.
+
+    The attribute, event and link it adds must survive into the finalized,
+    read-only span; otherwise the ``on_ending`` hook is running too late (after
+    the span was frozen).
+    """
+
+    def _on_ending(self, span: "trace.Span") -> None:
+        span.set_attribute("on_ending.attribute", "added")
+        span.add_event("on_ending.event")
+        span.add_link(
+            trace_api.SpanContext(1, 2, is_remote=False),
+            attributes={"on_ending.link": "added"},
+        )
+
+
 class TestSpanProcessor(unittest.TestCase):
     def test_span_processor(self):
         tracer_provider = trace.TracerProvider()
@@ -255,6 +272,75 @@ def test_on_ending_not_implemented_does_not_raise(self):
 
         self.assertListEqual(spans_calls_list, expected_list)
 
+    def test_on_ending_mutation_survives_synchronous(self):
+        # The default TracerProvider uses SynchronousMultiSpanProcessor.
+        tracer_provider = trace.TracerProvider()
+        tracer_provider.add_span_processor(MutatingOnEndingSpanProcessor())
+        exporter = InMemorySpanExporter()
+        tracer_provider.add_span_processor(SimpleSpanProcessor(exporter))
+        tracer = tracer_provider.get_tracer(__name__)
+
+        with tracer.start_as_current_span("foo"):
+            pass
+
+        finished_spans = exporter.get_finished_spans()
+        self.assertEqual(len(finished_spans), 1)
+        finished_span = finished_spans[0]
+
+        # The mutation performed inside on_ending must be reflected in the
+        # exported (read-only) span rather than silently dropped.
+        self.assertEqual(
+            finished_span.attributes.get("on_ending.attribute"), "added"
+        )
+        self.assertEqual(len(finished_span.events), 1)
+        self.assertEqual(finished_span.events[0].name, "on_ending.event")
+        self.assertEqual(len(finished_span.links), 1)
+        self.assertEqual(
+            finished_span.links[0].attributes.get("on_ending.link"), "added"
+        )
+
+    def test_on_ending_mutation_survives_concurrent(self):
+        concurrent_processor = trace.ConcurrentMultiSpanProcessor()
+        concurrent_processor.add_span_processor(
+            MutatingOnEndingSpanProcessor()
+        )
+        exporter = InMemorySpanExporter()
+        concurrent_processor.add_span_processor(SimpleSpanProcessor(exporter))
+
+        tracer_provider = trace.TracerProvider(
+            active_span_processor=concurrent_processor
+        )
+        tracer = tracer_provider.get_tracer(__name__)
+
+        with tracer.start_as_current_span("foo"):
+            pass
+
+        finished_spans = exporter.get_finished_spans()
+        self.assertEqual(len(finished_spans), 1)
+        finished_span = finished_spans[0]
+        self.assertEqual(
+            finished_span.attributes.get("on_ending.attribute"), "added"
+        )
+        self.assertEqual(len(finished_span.events), 1)
+        self.assertEqual(finished_span.events[0].name, "on_ending.event")
+        self.assertEqual(len(finished_span.links), 1)
+
+    def test_span_frozen_after_on_ending(self):
+        # A mutation attempted after the span has ended (outside on_ending)
+        # must still be rejected, i.e. the span is finalized once on_ending
+        # has completed.
+        tracer_provider = trace.TracerProvider()
+        tracer = tracer_provider.get_tracer(__name__)
+        span = tracer.start_span("bar")
+        span.end()
+
+        span.set_attribute("after_end.attribute", "rejected")
+        span.add_event("after_end.event")
+        self.assertNotIn("after_end.attribute", span.attributes)
+        self.assertFalse(
+            any(event.name == "after_end.event" for event in span.events)
+        )
+
 
 class MultiSpanProcessorTestBase(abc.ABC):
     @abc.abstractmethod