diff --git a/sentry_sdk/_span_batcher.py b/sentry_sdk/_span_batcher.py index deadf1b40e..79285c3386 100644 --- a/sentry_sdk/_span_batcher.py +++ b/sentry_sdk/_span_batcher.py @@ -21,16 +21,12 @@ class SpanBatcher(Batcher["SpanJSON"]): # MAX_BEFORE_FLUSH should be lower than MAX_BEFORE_DROP, so that there is # a bit of a buffer for spans that appear between the trigger to flush # and actually flushing the buffer. + # + # The max limits are all per trace (per bucket). MAX_ENVELOPE_SIZE = 1000 # spans - MAX_BEFORE_FLUSH = 1000 - GLOBAL_MAX_BEFORE_FLUSH = 5_000 - MAX_BEFORE_DROP = 2000 - GLOBAL_MAX_BEFORE_DROP = 10_000 - MAX_BYTES_BEFORE_FLUSH = 5 * 1024 * 1024 # 5 MB - GLOBAL_MAX_BYTES_BEFORE_FLUSH = 25 * 1024 * 1024 # 25 MB FLUSH_WAIT_TIME = 5.0 @@ -48,11 +44,7 @@ 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._total_running_size: int = 0 - self._capture_func = capture_func self._record_lost_func = record_lost_func self._running = True @@ -79,11 +71,7 @@ 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._total_running_size = 0 - self._running = True self._lock = threading.Lock() @@ -106,12 +94,8 @@ def _flush_loop(self) -> None: self._flush(only_pending=True) if ( - self._span_number >= self.GLOBAL_MAX_BEFORE_FLUSH - or self._total_running_size >= self.GLOBAL_MAX_BYTES_BEFORE_FLUSH - or ( - time.monotonic() - self._last_full_flush - >= self.FLUSH_WAIT_TIME + jitter - ) + time.monotonic() - self._last_full_flush + >= self.FLUSH_WAIT_TIME + jitter ): self._flush() self._last_full_flush = time.monotonic() @@ -132,10 +116,8 @@ def add(self, span: "SpanJSON") -> None: return None with self._lock: - if ( - self._span_number >= self.GLOBAL_MAX_BEFORE_DROP - or len(self._span_buffer[span["trace_id"]]) >= self.MAX_BEFORE_DROP - ): + size = len(self._span_buffer[span["trace_id"]]) + if size >= self.MAX_BEFORE_DROP: self._record_lost_func( reason="queue_overflow", data_category="span", @@ -144,25 +126,17 @@ def add(self, span: "SpanJSON") -> None: return None self._span_buffer[span["trace_id"]].append(span) - self._span_number += 1 - - estimated_size = self._estimate_size(span) - self._running_size[span["trace_id"]] += estimated_size - self._total_running_size += estimated_size + self._running_size[span["trace_id"]] += self._estimate_size(span) if ( - len(self._span_buffer[span["trace_id"]]) >= self.MAX_BEFORE_FLUSH + 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 = ( - self._span_number >= self.GLOBAL_MAX_BEFORE_FLUSH - or self._total_running_size - >= self.GLOBAL_MAX_BYTES_BEFORE_FLUSH - ) + notify = False if notify: self._flush_event.set() @@ -253,10 +227,7 @@ 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] - - self._total_running_size -= self._running_size[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 172117e742..c21d6cdea4 100644 --- a/tests/tracing/test_span_batcher.py +++ b/tests/tracing/test_span_batcher.py @@ -171,88 +171,6 @@ 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) @@ -338,130 +256,6 @@ def test_weight_based_flushing_by_attribute_size( assert envelopes[0].items[0].payload.json["items"][1]["name"] == "big span" -def test_global_length_based_flushing(sentry_init, capture_items, monkeypatch): - """A flush event is triggered when the batcher contains GLOBAL_MAX_BEFORE_FLUSH spans.""" - monkeypatch.setattr(SpanBatcher, "GLOBAL_MAX_BEFORE_FLUSH", 2) - # set the time-based flush limit to something huge so that we're not hitting - # it since we want to test GLOBAL_MAX_BEFORE_FLUSH instead - monkeypatch.setattr(SpanBatcher, "FLUSH_WAIT_TIME", 100000) - - sentry_init( - traces_sample_rate=1.0, - trace_lifecycle="stream", - ) - - items = capture_items("span") - - with sentry_sdk.traces.start_span(name="span"): - pass - - sentry_sdk.traces.new_trace() - with sentry_sdk.traces.start_span(name="span 2"): - pass - - time.sleep(0.1) - - assert len(items) == 2 - assert items[0].payload["name"] == "span" - - -def test_span_number_reset_after_length_based_flushing( - sentry_init, capture_items, monkeypatch -): - """Span is not flushed after a flush reduces the number of spans in the batcher below the global limit.""" - monkeypatch.setattr(SpanBatcher, "GLOBAL_MAX_BEFORE_FLUSH", 2) - # set the time-based flush limit to something huge so that we're not hitting - # it since we want to test GLOBAL_MAX_BYTES_BEFORE_FLUSH instead - monkeypatch.setattr(SpanBatcher, "FLUSH_WAIT_TIME", 100000) - - sentry_init( - traces_sample_rate=1.0, - trace_lifecycle="stream", - ) - - items = capture_items("span") - - with sentry_sdk.traces.start_span(name="span"): - pass - - sentry_sdk.traces.new_trace() - with sentry_sdk.traces.start_span(name="span"): - pass - - time.sleep(0.1) - - with sentry_sdk.traces.start_span(name="span"): - pass - - time.sleep(0.1) - - assert len(items) == 2 - assert items[0].payload["name"] == "span" - - -def test_global_weight_based_flushing(sentry_init, capture_items, monkeypatch): - """When the batcher reaches GLOBAL_MAX_BYTES_BEFORE_FLUSH, all buckets will be flushed.""" - # Limit of 2_000 is just above the size of a bare span. - monkeypatch.setattr(SpanBatcher, "GLOBAL_MAX_BYTES_BEFORE_FLUSH", 2_000) - # set the time-based flush limit to something huge so that it doesn't - # interfere - monkeypatch.setattr(SpanBatcher, "FLUSH_WAIT_TIME", 100000) - - sentry_init( - traces_sample_rate=1.0, - trace_lifecycle="stream", - ) - - items = capture_items("span") - - with sentry_sdk.traces.start_span(name="span"): - pass - - sentry_sdk.traces.new_trace() - with sentry_sdk.traces.start_span(name="span"): - pass - - time.sleep(0.1) - - assert len(items) == 2 - assert items[0].payload["name"] == "span" - - -def test_total_size_reset_after_weight_based_flushing( - sentry_init, capture_items, monkeypatch -): - """Span is not flushed after a flush reduces the combined span size in bytes below the global limit.""" - # Limit of 2_000 is just above the size of a bare span. - monkeypatch.setattr(SpanBatcher, "GLOBAL_MAX_BYTES_BEFORE_FLUSH", 2_000) - # set the time-based flush limit to something huge so that it doesn't - # interfere - monkeypatch.setattr(SpanBatcher, "FLUSH_WAIT_TIME", 100000) - - sentry_init( - traces_sample_rate=1.0, - trace_lifecycle="stream", - ) - - items = capture_items("span") - - with sentry_sdk.traces.start_span(name="span"): - pass - - sentry_sdk.traces.new_trace() - with sentry_sdk.traces.start_span(name="span"): - pass - - time.sleep(0.1) - - with sentry_sdk.traces.start_span(name="span"): - pass - - time.sleep(0.1) - - assert len(items) == 2 - assert items[0].payload["name"] == "span" - - def test_bucket_recreated_after_flush(sentry_init, capture_envelopes, monkeypatch): """Spans for a trace that arrive after that trace's bucket was flushed land in a fresh bucket.""" monkeypatch.setattr(SpanBatcher, "MAX_BEFORE_FLUSH", 2) @@ -666,11 +460,7 @@ 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._total_running_size = 42 - batcher._active.flag = True batcher._flush_event.set() batcher._running = False @@ -682,10 +472,7 @@ 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 - total_running_size_reset = batcher._total_running_size == 0 active_reset = not getattr(batcher._active, "flag", False) event_reset = not batcher._flush_event.is_set() @@ -697,9 +484,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 total_running_size_reset and active_reset and event_reset and running_reset