diff --git a/.changelog/14.fixed b/.changelog/14.fixed new file mode 100644 index 0000000000..7a3e296527 --- /dev/null +++ b/.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 diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py index fe800c3695..2560cca823 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py +++ b/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,11 +902,23 @@ 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 @@ -898,7 +926,7 @@ def set_attributes( 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 diff --git a/opentelemetry-sdk/tests/trace/test_span_processor.py b/opentelemetry-sdk/tests/trace/test_span_processor.py index 4499427cc1..e4aadc171a 100644 --- a/opentelemetry-sdk/tests/trace/test_span_processor.py +++ b/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