diff --git a/sentry_sdk/_span_batcher.py b/sentry_sdk/_span_batcher.py index ea67ff6e01..f6c0624568 100644 --- a/sentry_sdk/_span_batcher.py +++ b/sentry_sdk/_span_batcher.py @@ -127,26 +127,30 @@ def add(self, span: "SpanJSON") -> None: with self._lock: size = len(self._span_buffer[span["trace_id"]]) if size >= self.MAX_BEFORE_DROP: - self._record_lost_func( - reason="queue_overflow", - data_category="span", - quantity=1, - ) - return None - - self._span_buffer[span["trace_id"]].append(span) - self._running_size[span["trace_id"]] += self._estimate_size(span) - - if ( - span["is_segment"] is True - or size + 1 >= self.MAX_BEFORE_FLUSH - or self._running_size[span["trace_id"]] - >= self.MAX_BYTES_BEFORE_FLUSH - ): - self._pending_flush.add(span["trace_id"]) - notify = True + dropped = True else: - notify = False + dropped = False + self._span_buffer[span["trace_id"]].append(span) + self._running_size[span["trace_id"]] += self._estimate_size(span) + + if ( + span["is_segment"] is True + or size + 1 >= self.MAX_BEFORE_FLUSH + or self._running_size[span["trace_id"]] + >= self.MAX_BYTES_BEFORE_FLUSH + ): + self._pending_flush.add(span["trace_id"]) + notify = True + else: + notify = False + + if dropped: + self._record_lost_func( + reason="queue_overflow", + data_category="span", + quantity=1, + ) + return None if notify: self._flush_event.set() @@ -195,50 +199,55 @@ def _flush(self, only_pending: bool = False) -> None: if not buckets: return - envelopes = [] - + flushed_buckets: list[tuple[str, list["SpanJSON"]]] = [] for bucket_id in buckets: - spans = self._span_buffer.get(bucket_id) - if not spans: - continue - - dsc = spans[0]["_segment_span"]._dynamic_sampling_context() - - # Max per envelope is 1000, so if we happen to have more than - # 1000 spans in one bucket, we'll need to separate them. - for start in range(0, len(spans), self.MAX_ENVELOPE_SIZE): - end = min(start + self.MAX_ENVELOPE_SIZE, len(spans)) - - envelope = Envelope( + spans = self._span_buffer.pop(bucket_id, None) + if spans: + flushed_buckets.append((bucket_id, spans)) + self._running_size.pop(bucket_id, None) + + if not flushed_buckets: + return + + envelopes = [] + + # Serialize outside the lock to prevent blocking concurrent additions + # and avoid lock-order inversion during serialization or GC finalizers. + for _, spans in flushed_buckets: + dsc = spans[0]["_segment_span"]._dynamic_sampling_context() + + # Max per envelope is 1000, so if we happen to have more than + # 1000 spans in one bucket, we'll need to separate them. + for start in range(0, len(spans), self.MAX_ENVELOPE_SIZE): + end = min(start + self.MAX_ENVELOPE_SIZE, len(spans)) + + envelope = Envelope( + headers={ + "sent_at": format_timestamp(datetime.now(timezone.utc)), + "trace": dsc, + } + ) + + envelope.add_item( + Item( + type=self.TYPE, + content_type=self.CONTENT_TYPE, headers={ - "sent_at": format_timestamp(datetime.now(timezone.utc)), - "trace": dsc, - } + "item_count": end - start, + }, + payload=PayloadRef( + json={ + "version": 2, + "items": [ + self._to_transport_format(spans[j]) + for j in range(start, end) + ], + } + ), ) + ) - envelope.add_item( - Item( - type=self.TYPE, - content_type=self.CONTENT_TYPE, - headers={ - "item_count": end - start, - }, - payload=PayloadRef( - json={ - "version": 2, - "items": [ - self._to_transport_format(spans[j]) - for j in range(start, end) - ], - } - ), - ) - ) - - envelopes.append(envelope) - - del self._span_buffer[bucket_id] - del self._running_size[bucket_id] + envelopes.append(envelope) for envelope in envelopes: self._capture_func(envelope) diff --git a/tests/tracing/test_span_batcher.py b/tests/tracing/test_span_batcher.py index f249125c90..485ba59c6d 100644 --- a/tests/tracing/test_span_batcher.py +++ b/tests/tracing/test_span_batcher.py @@ -572,3 +572,112 @@ def _flush(self, only_pending=False): batcher._flush_loop() assert calls == [1] + + +def test_flush_serializes_outside_the_lock(sentry_init, capture_envelopes): + """Regression test for https://github.com/getsentry/sentry-python/issues/8023.""" + sentry_init( + traces_sample_rate=1.0, + trace_lifecycle="stream", + ) + capture_envelopes() + + client = sentry_sdk.get_client() + batcher = client.span_batcher + + lock_held_while_serializing = [] + original_to_transport_format = batcher._to_transport_format + + def record_lock_state(item): + lock_held_while_serializing.append(batcher._lock.locked()) + return original_to_transport_format(item) + + batcher._to_transport_format = record_lock_state + + with sentry_sdk.traces.start_span(name="test span"): + pass + + client.flush() + + assert lock_held_while_serializing + assert not any(lock_held_while_serializing) + + +def test_concurrent_add_not_blocked_during_serialization( + sentry_init, capture_envelopes +): + """Verify concurrent additions are not blocked while spans are serialized.""" + import threading + + sentry_init( + traces_sample_rate=1.0, + trace_lifecycle="stream", + ) + capture_envelopes() + + client = sentry_sdk.get_client() + batcher = client.span_batcher + + serializing_event = threading.Event() + continue_event = threading.Event() + original_to_transport_format = batcher._to_transport_format + + def pause_serializing(item): + serializing_event.set() + continue_event.wait(timeout=5) + return original_to_transport_format(item) + + batcher._to_transport_format = pause_serializing + + with sentry_sdk.traces.start_span(name="span 1"): + pass + + flush_thread = threading.Thread(target=client.flush) + flush_thread.start() + + assert serializing_event.wait(timeout=5) + + # The lock must NOT be held while serialization is paused + lock_acquired = batcher._lock.acquire(blocking=False) + if lock_acquired: + batcher._lock.release() + + continue_event.set() + flush_thread.join() + + assert lock_acquired is True + + +def test_record_lost_called_outside_lock( + sentry_init, capture_record_lost_event_calls, monkeypatch +): + """Verify record_lost is called outside the batcher lock when buffer overflows.""" + monkeypatch.setattr(SpanBatcher, "MAX_BEFORE_DROP", 1) + monkeypatch.setattr(SpanBatcher, "FLUSH_WAIT_TIME", 100000) + + sentry_init( + traces_sample_rate=1.0, + trace_lifecycle="stream", + ) + record_lost_event_calls = capture_record_lost_event_calls() + client = sentry_sdk.get_client() + batcher = client.span_batcher + + lock_held_on_lost = [] + original_record_lost = batcher._record_lost_func + + def record_lost(*args, **kwargs): + lock_held_on_lost.append(batcher._lock.locked()) + return original_record_lost(*args, **kwargs) + + batcher._record_lost_func = record_lost + + with sentry_sdk.traces.start_span(name="parent"): + with sentry_sdk.traces.start_span(name="span 1"): + pass + with sentry_sdk.traces.start_span(name="span 2"): + pass + + assert all(not held for held in lock_held_on_lost) + assert len(lock_held_on_lost) >= 1 + assert ("queue_overflow", "span", None, 1) in record_lost_event_calls