Skip to content

ref: Revert global batcher limilts #7168

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
alexander-alderman-webb wants to merge 3 commits into
base: master
Choose a base branch
from
+9 −253
Open

ref: Revert global batcher limilts #7168

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0414de5
Revert "fix(batcher): Add global flush trigger based on the span numb…
alexander-alderman-webb Aug 11, 2026
dd57e21
Revert "feat(batcher): Add global flush trigger based on summed size …
alexander-alderman-webb Aug 11, 2026
8669777
Revert "feat(batcher): Add hard span limit (#7143)"
alexander-alderman-webb Aug 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 9 additions & 38 deletions sentry_sdk/_span_batcher.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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
Expand All @@ -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()
Expand All @@ -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()
Expand All @@ -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",
Expand All @@ -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()
Expand Down Expand Up @@ -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:
Expand Down
215 changes: 0 additions & 215 deletions tests/tracing/test_span_batcher.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand All @@ -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()
Expand All @@ -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
Expand Down
Loading