diff --git a/sentry_sdk/_span_batcher.py b/sentry_sdk/_span_batcher.py index 79285c3386..c1c40044ee 100644 --- a/sentry_sdk/_span_batcher.py +++ b/sentry_sdk/_span_batcher.py @@ -25,7 +25,10 @@ class SpanBatcher(Batcher["SpanJSON"]): # The max limits are all per trace (per bucket). MAX_ENVELOPE_SIZE = 1000 # spans MAX_BEFORE_FLUSH = 1000 + MAX_BEFORE_DROP = 2000 + GLOBAL_MAX_BEFORE_DROP = 10_000 + MAX_BYTES_BEFORE_FLUSH = 5 * 1024 * 1024 # 5 MB FLUSH_WAIT_TIME = 5.0 @@ -44,6 +47,8 @@ def __init__( # envelope. # trace_id -> span buffer self._span_buffer: dict[str, list["SpanJSON"]] = defaultdict(list) + self._span_number: int = 0 + self._running_size: dict[str, int] = defaultdict(lambda: 0) self._capture_func = capture_func self._record_lost_func = record_lost_func @@ -71,6 +76,8 @@ def _reset_in_child() -> None: def _reset_thread_state(self) -> None: self._span_buffer = defaultdict(list) + self._span_number = 0 + self._running_size = defaultdict(lambda: 0) self._running = True @@ -116,8 +123,10 @@ def add(self, span: "SpanJSON") -> None: return None with self._lock: - size = len(self._span_buffer[span["trace_id"]]) - if size >= self.MAX_BEFORE_DROP: + if ( + self._span_number >= self.GLOBAL_MAX_BEFORE_DROP + or len(self._span_buffer[span["trace_id"]]) >= self.MAX_BEFORE_DROP + ): self._record_lost_func( reason="queue_overflow", data_category="span", @@ -126,10 +135,12 @@ def add(self, span: "SpanJSON") -> None: return None self._span_buffer[span["trace_id"]].append(span) + self._span_number += 1 + self._running_size[span["trace_id"]] += self._estimate_size(span) if ( - size + 1 >= self.MAX_BEFORE_FLUSH + len(self._span_buffer[span["trace_id"]]) >= self.MAX_BEFORE_FLUSH or self._running_size[span["trace_id"]] >= self.MAX_BYTES_BEFORE_FLUSH ): @@ -227,7 +238,9 @@ def _flush(self, only_pending: bool = False) -> None: envelopes.append(envelope) + self._span_number -= len(self._span_buffer[bucket_id]) del self._span_buffer[bucket_id] + del self._running_size[bucket_id] for envelope in envelopes: diff --git a/tests/tracing/test_span_batcher.py b/tests/tracing/test_span_batcher.py index c21d6cdea4..507fef184b 100644 --- a/tests/tracing/test_span_batcher.py +++ b/tests/tracing/test_span_batcher.py @@ -171,6 +171,88 @@ def test_drop_isolated_per_bucket( assert record_lost_event_calls.count(("queue_overflow", "span", None, 1)) == 1 +def test_drop_after_global_max_reached( + sentry_init, capture_envelopes, capture_record_lost_event_calls, monkeypatch +): + """New spans are dropped if the buffer reaches GLOBAL_MAX_BEFORE_DROP spans.""" + monkeypatch.setattr(SpanBatcher, "GLOBAL_MAX_BEFORE_DROP", 2) + # set the time-based flush limit to something huge so that we're not flushing + # prematurely + monkeypatch.setattr(SpanBatcher, "FLUSH_WAIT_TIME", 100000) + + sentry_init( + traces_sample_rate=1.0, + trace_lifecycle="stream", + ) + + envelopes = capture_envelopes() + record_lost_event_calls = capture_record_lost_event_calls() + + with sentry_sdk.traces.start_span(name="span 1"): + pass + with sentry_sdk.traces.start_span(name="span 2"): + pass + with sentry_sdk.traces.start_span(name="span 3"): + pass + + sentry_sdk.traces.new_trace() + with sentry_sdk.traces.start_span(name="span 4"): + pass + + sentry_sdk.flush() + + assert len(envelopes) == 1 + + assert len(envelopes[0].items[0].payload.json["items"]) == 2 + assert envelopes[0].items[0].payload.json["items"][0]["name"] == "span 1" + assert envelopes[0].items[0].payload.json["items"][1]["name"] == "span 2" + + assert record_lost_event_calls.count(("queue_overflow", "span", None, 1)) == 2 + + +def test_capture_after_flush_with_global_limit( + sentry_init, capture_envelopes, monkeypatch +): + """New spans are captured again after a flush reduces the span number below the global limit.""" + monkeypatch.setattr(SpanBatcher, "GLOBAL_MAX_BEFORE_DROP", 2) + # set the time-based flush limit to something huge so that we're not flushing + # prematurely + monkeypatch.setattr(SpanBatcher, "FLUSH_WAIT_TIME", 100000) + + sentry_init( + traces_sample_rate=1.0, + trace_lifecycle="stream", + ) + + envelopes = capture_envelopes() + + with sentry_sdk.traces.start_span(name="span 1"): + pass + with sentry_sdk.traces.start_span(name="span 2"): + pass + + sentry_sdk.traces.new_trace() + with sentry_sdk.traces.start_span(name="span 3"): + pass + + sentry_sdk.flush() + + # The span is captured even though a span was dropped in the same trace. + with sentry_sdk.traces.start_span(name="span 4"): + pass + + sentry_sdk.flush() + + assert len(envelopes) == 2 + + assert len(envelopes[0].items[0].payload.json["items"]) == 2 + assert envelopes[0].items[0].payload.json["items"][0]["name"] == "span 1" + assert envelopes[0].items[0].payload.json["items"][1]["name"] == "span 2" + + assert len(envelopes[1].items[0].payload.json["items"]) == 1 + assert envelopes[1].items[0].payload.json["items"][0]["name"] == "span 4" + + def test_length_based_flushing(sentry_init, capture_items, monkeypatch): """A flush event is triggered when a bucket contains MAX_BEFORE_FLUSH spans.""" monkeypatch.setattr(SpanBatcher, "MAX_BEFORE_FLUSH", 1) @@ -460,6 +542,8 @@ def test_span_batcher_lock_reset_in_child_after_fork(sentry_init): original_lock.acquire() batcher._span_buffer["test-trace-id"].append(object()) + batcher._span_number = 1 + batcher._running_size["test-trace-id"] = 42 batcher._active.flag = True batcher._flush_event.set() @@ -472,6 +556,8 @@ def test_span_batcher_lock_reset_in_child_after_fork(sentry_init): flusher_reset = batcher._flusher is None and batcher._flusher_pid is None span_buffer_reset = len(batcher._span_buffer) == 0 + span_number_reset = batcher._span_number == 0 + running_size_reset = len(batcher._running_size) == 0 active_reset = not getattr(batcher._active, "flag", False) @@ -484,6 +570,7 @@ def test_span_batcher_lock_reset_in_child_after_fork(sentry_init): and unheld and flusher_reset and span_buffer_reset + and span_number_reset and running_size_reset and active_reset and event_reset