ref: Revert global span batcher limits (#7168)

Author: alexander-alderman-webb

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:

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