diff --git a/sentry_sdk/_span_batcher.py b/sentry_sdk/_span_batcher.py index 79285c3386..c343b533eb 100644 --- a/sentry_sdk/_span_batcher.py +++ b/sentry_sdk/_span_batcher.py @@ -129,7 +129,8 @@ def add(self, span: "SpanJSON") -> None: self._running_size[span["trace_id"]] += self._estimate_size(span) if ( - size + 1 >= self.MAX_BEFORE_FLUSH + span["is_segment"] is True + or size + 1 >= self.MAX_BEFORE_FLUSH or self._running_size[span["trace_id"]] >= self.MAX_BYTES_BEFORE_FLUSH ): diff --git a/tests/conftest.py b/tests/conftest.py index 599c075224..312c992e6b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,6 +4,7 @@ import json import os import socket +import threading import warnings from collections import namedtuple from contextlib import contextmanager @@ -293,11 +294,71 @@ def inner(identifier): return inner +def _install_flush_completion_handshake(client: "sentry_sdk.Client") -> None: + """Make batcher.flush() wait for the flusher thread to drain. + + Otherwise, test assertions can run before envelopes are captured. + The span batcher flushes pending items asynchronously with the main thread. + Flushes triggered by segments finishing are asynchronous, and can collect buckets + that would have otherwise been flushed synchronously by `sentry_sdk.flush()`. + """ + batcher = client.span_batcher + if batcher is None: + return + + orig_flush_raw = batcher._flush + orig_flush = batcher.flush + lock = threading.Lock() + drained_count = 0 + wake = threading.Event() + + def _flush(*args: "Any", **kwargs: "Any") -> "Any": + nonlocal drained_count + try: + return orig_flush_raw(*args, **kwargs) + finally: + with lock: + drained_count += 1 + wake.set() + + def flush() -> None: + nonlocal drained_count + # Re-entrancy guard: if `flush()` is invoked from within a drain (e.g. a + # custom transport), waiting on the flusher thread would deadlock, because + # the flusher is blocked inside our own handler. + if getattr(getattr(batcher, "_active", None), "flag", False): + orig_flush() + return + + # If the background flusher thread was never started (no spans have + # been added), there is no thread to drain and the counter will never + # advance. Fall back to the original synchronous flush. + if batcher._flusher is None or not batcher._flusher.is_alive(): + orig_flush() + return + + with lock: + target = drained_count + + batcher._flush_event.set() + while True: + with lock: + if drained_count > target: + break + wake.wait() + wake.clear() + orig_flush() + + object.__setattr__(batcher, "_flush", _flush) + object.__setattr__(batcher, "flush", flush) + + @pytest.fixture def sentry_init(request): def inner(*a, **kw): kw.setdefault("transport", TestTransport()) client = sentry_sdk.Client(*a, **kw) + _install_flush_completion_handshake(client) sentry_sdk.get_global_scope().set_client(client) if request.node.get_closest_marker("forked"): diff --git a/tests/integrations/django/test_basic.py b/tests/integrations/django/test_basic.py index 03e62fe6ec..0b027a7b83 100644 --- a/tests/integrations/django/test_basic.py +++ b/tests/integrations/django/test_basic.py @@ -2499,12 +2499,17 @@ def test_transaction_http_method_custom( client.get("/nomessage") client.options("/nomessage") - client.head("/nomessage") sentry_sdk.flush() spans = [item.payload for item in items] assert spans[2]["attributes"][SPANDATA.HTTP_REQUEST_METHOD] == "OPTIONS" + + client.head("/nomessage") + + sentry_sdk.flush() + spans = [item.payload for item in items] + assert spans[5]["attributes"][SPANDATA.HTTP_REQUEST_METHOD] == "HEAD" else: events = capture_events() diff --git a/tests/integrations/django/test_cache_module.py b/tests/integrations/django/test_cache_module.py index b00903104d..700c5e3a80 100644 --- a/tests/integrations/django/test_cache_module.py +++ b/tests/integrations/django/test_cache_module.py @@ -246,7 +246,6 @@ def test_cache_spans_middleware( if span_streaming: items = capture_items("span") - client.get(reverse("not_cached_view")) client.get(reverse("not_cached_view")) sentry_sdk.flush() @@ -269,6 +268,11 @@ def test_cache_spans_middleware( ) assert "cache.hit" not in spans[1]["attributes"] assert spans[1]["attributes"]["cache.item_size"] == 2 + + client.get(reverse("not_cached_view")) + + sentry_sdk.flush() + spans = [item.payload for item in items] # second_event - cache.get assert spans[4]["attributes"]["sentry.op"] == "cache.get" assert spans[4]["name"].startswith("views.decorators.cache.cache_header.") diff --git a/tests/integrations/flask/test_flask.py b/tests/integrations/flask/test_flask.py index cadf011b4c..1252673b34 100644 --- a/tests/integrations/flask/test_flask.py +++ b/tests/integrations/flask/test_flask.py @@ -1211,20 +1211,31 @@ def test_transaction_or_segment_http_method_custom( response = client.options("/nomessage") assert response.status_code == 200 - response = client.head("/nomessage") - assert response.status_code == 200 - if span_streaming: sentry_sdk.flush() spans = [i.payload for i in items] - assert len(spans) == 2 - (options_segment, head_segment) = spans + (options_segment,) = spans assert options_segment["attributes"]["http.request.method"] == "OPTIONS" + + response = client.head("/nomessage") + assert response.status_code == 200 + + sentry_sdk.flush() + spans = [i.payload for i in items] + assert len(spans) == 2 + (_, head_segment) = spans + assert head_segment["attributes"]["http.request.method"] == "HEAD" else: - assert len(events) == 2 - (event1, event2) = events + (event1,) = events assert event1["request"]["method"] == "OPTIONS" + + response = client.head("/nomessage") + assert response.status_code == 200 + + assert len(events) == 2 + (_, event2) = events + assert event2["request"]["method"] == "HEAD" diff --git a/tests/tracing/test_span_batcher.py b/tests/tracing/test_span_batcher.py index c21d6cdea4..679a2be4a7 100644 --- a/tests/tracing/test_span_batcher.py +++ b/tests/tracing/test_span_batcher.py @@ -18,19 +18,27 @@ def test_envelope_by_trace_id(sentry_init, capture_envelopes, monkeypatch): envelopes = capture_envelopes() - sentry_sdk.traces.new_trace() - - with sentry_sdk.traces.start_span(name="span 1a") as span1: - trace_id1 = span1.trace_id - with sentry_sdk.traces.start_span(name="span 1b"): - pass - - sentry_sdk.traces.new_trace() + with sentry_sdk.new_scope(): + sentry_sdk.traces.new_trace() + # Keep parent open as its bucket in the batcher would be emptied when it is finished. + parent_a = sentry_sdk.traces.start_span(name="parent a") + with sentry_sdk.traces.start_span( + name="span 1a", parent_span=parent_a + ) as span1: + trace_id1 = span1.trace_id + with sentry_sdk.traces.start_span(name="span 1b", parent_span=parent_a): + pass - with sentry_sdk.traces.start_span(name="span 2a") as span2: - trace_id2 = span2.trace_id - with sentry_sdk.traces.start_span(name="span 2b"): - pass + with sentry_sdk.new_scope(): + sentry_sdk.traces.new_trace() + parent_b = sentry_sdk.traces.start_span(name="parent b") + # Keep parent open as its bucket in the batcher would be emptied when it is finished. + with sentry_sdk.traces.start_span( + name="span 2a", parent_span=parent_b + ) as span2: + trace_id2 = span2.trace_id + with sentry_sdk.traces.start_span(name="span 2b", parent_span=parent_b): + pass sentry_sdk.flush() @@ -65,29 +73,30 @@ def test_max_envelope_size(sentry_init, capture_envelopes, monkeypatch): envelopes = capture_envelopes() - 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 - with sentry_sdk.traces.start_span(name="span 4"): - pass - with sentry_sdk.traces.start_span(name="span 5"): - pass + with sentry_sdk.traces.start_span(name="custom parent"): + 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 + with sentry_sdk.traces.start_span(name="span 4"): + pass + with sentry_sdk.traces.start_span(name="span 5"): + pass - sentry_sdk.flush() + sentry_sdk.flush() - assert len(envelopes) == 3 + assert len(envelopes) == 3 - 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"]) == 2 - assert envelopes[1].items[0].payload.json["items"][0]["name"] == "span 3" - assert envelopes[1].items[0].payload.json["items"][1]["name"] == "span 4" - assert len(envelopes[2].items[0].payload.json["items"]) == 1 - assert envelopes[2].items[0].payload.json["items"][0]["name"] == "span 5" + 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"]) == 2 + assert envelopes[1].items[0].payload.json["items"][0]["name"] == "span 3" + assert envelopes[1].items[0].payload.json["items"][1]["name"] == "span 4" + assert len(envelopes[2].items[0].payload.json["items"]) == 1 + assert envelopes[2].items[0].payload.json["items"][0]["name"] == "span 5" def test_drop_after_max_reached( @@ -107,22 +116,23 @@ def test_drop_after_max_reached( 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 + with sentry_sdk.traces.start_span(name="custom parent"): + 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.flush() + sentry_sdk.flush() - assert len(envelopes) == 1 + 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 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 ("queue_overflow", "span", None, 1) in record_lost_event_calls + assert ("queue_overflow", "span", None, 1) in record_lost_event_calls def test_drop_isolated_per_bucket( @@ -140,19 +150,24 @@ def test_drop_isolated_per_bucket( envelopes = capture_envelopes() record_lost_event_calls = capture_record_lost_event_calls() - sentry_sdk.traces.new_trace() - with sentry_sdk.traces.start_span(name="a1") as span_a: - trace_id_a = span_a.trace_id - with sentry_sdk.traces.start_span(name="a2"): - pass - with sentry_sdk.traces.start_span(name="a3"): - pass + with sentry_sdk.new_scope(): + # Keep parent open as its bucket in the batcher would be emptied when it is finished. + parent_a = sentry_sdk.traces.start_span(name="parent a") + with sentry_sdk.traces.start_span(name="a1", parent_span=parent_a) as span_a: + trace_id_a = span_a.trace_id + with sentry_sdk.traces.start_span(name="a2", parent_span=parent_a): + pass + with sentry_sdk.traces.start_span(name="a3"): + pass - sentry_sdk.traces.new_trace() - with sentry_sdk.traces.start_span(name="b1") as span_b: - trace_id_b = span_b.trace_id - with sentry_sdk.traces.start_span(name="b2"): - pass + with sentry_sdk.new_scope(): + sentry_sdk.traces.new_trace() + # Keep parent open as its bucket in the batcher would be emptied when it is finished. + parent_b = sentry_sdk.traces.start_span(name="parent b") + with sentry_sdk.traces.start_span(name="b1", parent_span=parent_b) as span_b: + trace_id_b = span_b.trace_id + with sentry_sdk.traces.start_span(name="b2", parent_span=parent_b): + pass sentry_sdk.flush() @@ -185,13 +200,14 @@ def test_length_based_flushing(sentry_init, capture_items, monkeypatch): items = capture_items("span") - with sentry_sdk.traces.start_span(name="span"): - pass + with sentry_sdk.traces.start_span(name="custom parent"): + with sentry_sdk.traces.start_span(name="span"): + pass - time.sleep(0.1) + time.sleep(0.1) - assert len(items) == 1 - assert items[0].payload["name"] == "span" + assert len(items) == 1 + assert items[0].payload["name"] == "span" def test_weight_based_flushing(sentry_init, capture_envelopes, monkeypatch): @@ -208,15 +224,16 @@ def test_weight_based_flushing(sentry_init, capture_envelopes, monkeypatch): envelopes = capture_envelopes() - with sentry_sdk.traces.start_span(name="span"): - pass + with sentry_sdk.traces.start_span(name="custom parent"): + with sentry_sdk.traces.start_span(name="span"): + pass - time.sleep(0.1) + time.sleep(0.1) - assert len(envelopes) == 1 + assert len(envelopes) == 1 - assert len(envelopes[0].items[0].payload.json["items"]) == 1 - assert envelopes[0].items[0].payload.json["items"][0]["name"] == "span" + assert len(envelopes[0].items[0].payload.json["items"]) == 1 + assert envelopes[0].items[0].payload.json["items"][0]["name"] == "span" def test_weight_based_flushing_by_attribute_size( @@ -233,27 +250,30 @@ def test_weight_based_flushing_by_attribute_size( envelopes = capture_envelopes() - with sentry_sdk.traces.start_span(name="small span") as bare_span: - pass + with sentry_sdk.traces.start_span(name="custom parent"): + with sentry_sdk.traces.start_span(name="small span") as bare_span: + pass - bare_span_size = SpanBatcher._estimate_size(bare_span._to_json()) - big_attr = "x" * bare_span_size + bare_span_size = SpanBatcher._estimate_size(bare_span._to_json()) + big_attr = "x" * bare_span_size - monkeypatch.setattr(SpanBatcher, "MAX_BYTES_BEFORE_FLUSH", bare_span_size * 3) + monkeypatch.setattr(SpanBatcher, "MAX_BYTES_BEFORE_FLUSH", bare_span_size * 3) - time.sleep(0.1) + time.sleep(0.1) - # The first span alone is well under the byte limit, so no flush yet. - assert len(envelopes) == 0 + # The first span alone is well under the byte limit, so no flush yet. + assert len(envelopes) == 0 - with sentry_sdk.traces.start_span(name="big span", attributes={"big": big_attr}): - pass + with sentry_sdk.traces.start_span( + name="big span", attributes={"big": big_attr} + ): + pass - time.sleep(0.1) + time.sleep(0.1) - assert len(envelopes) == 1 - assert envelopes[0].items[0].payload.json["items"][0]["name"] == "small span" - assert envelopes[0].items[0].payload.json["items"][1]["name"] == "big span" + assert len(envelopes) == 1 + assert envelopes[0].items[0].payload.json["items"][0]["name"] == "small span" + assert envelopes[0].items[0].payload.json["items"][1]["name"] == "big span" def test_bucket_recreated_after_flush(sentry_init, capture_envelopes, monkeypatch): @@ -270,37 +290,38 @@ def test_bucket_recreated_after_flush(sentry_init, capture_envelopes, monkeypatc sentry_sdk.traces.new_trace() - with sentry_sdk.traces.start_span(name="span 1") as span1: - trace_id = span1.trace_id - with sentry_sdk.traces.start_span(name="span 2"): - pass + with sentry_sdk.traces.start_span(name="custom parent"): + with sentry_sdk.traces.start_span(name="span 1") as span1: + trace_id = span1.trace_id + with sentry_sdk.traces.start_span(name="span 2"): + pass - time.sleep(0.1) + time.sleep(0.1) - assert len(envelopes) == 1 + assert len(envelopes) == 1 - with sentry_sdk.traces.start_span(name="span 3"): - pass - with sentry_sdk.traces.start_span(name="span 4"): - pass + with sentry_sdk.traces.start_span(name="span 3"): + pass + with sentry_sdk.traces.start_span(name="span 4"): + pass - time.sleep(0.1) + time.sleep(0.1) - assert len(envelopes) == 2 + assert len(envelopes) == 2 - assert envelopes[0].headers["trace"]["trace_id"] == trace_id - 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"][0]["trace_id"] == trace_id - assert envelopes[0].items[0].payload.json["items"][1]["name"] == "span 2" - assert envelopes[0].items[0].payload.json["items"][1]["trace_id"] == trace_id + assert envelopes[0].headers["trace"]["trace_id"] == trace_id + 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"][0]["trace_id"] == trace_id + assert envelopes[0].items[0].payload.json["items"][1]["name"] == "span 2" + assert envelopes[0].items[0].payload.json["items"][1]["trace_id"] == trace_id - assert envelopes[1].headers["trace"]["trace_id"] == trace_id - assert len(envelopes[1].items[0].payload.json["items"]) == 2 - assert envelopes[1].items[0].payload.json["items"][0]["name"] == "span 3" - assert envelopes[1].items[0].payload.json["items"][0]["trace_id"] == trace_id - assert envelopes[1].items[0].payload.json["items"][1]["name"] == "span 4" - assert envelopes[1].items[0].payload.json["items"][1]["trace_id"] == trace_id + assert envelopes[1].headers["trace"]["trace_id"] == trace_id + assert len(envelopes[1].items[0].payload.json["items"]) == 2 + assert envelopes[1].items[0].payload.json["items"][0]["name"] == "span 3" + assert envelopes[1].items[0].payload.json["items"][0]["trace_id"] == trace_id + assert envelopes[1].items[0].payload.json["items"][1]["name"] == "span 4" + assert envelopes[1].items[0].payload.json["items"][1]["trace_id"] == trace_id def test_quiet_buckets_flush_eventually(sentry_init, capture_envelopes, monkeypatch): @@ -317,15 +338,16 @@ def test_quiet_buckets_flush_eventually(sentry_init, capture_envelopes, monkeypa envelopes = capture_envelopes() - with sentry_sdk.traces.start_span(name="span 1"): - pass + with sentry_sdk.traces.start_span(name="custom parent"): + with sentry_sdk.traces.start_span(name="span 1"): + pass - time.sleep(0.3) + time.sleep(0.3) - assert len(envelopes) == 1 + assert len(envelopes) == 1 - assert len(envelopes[0].items[0].payload.json["items"]) == 1 - assert envelopes[0].items[0].payload.json["items"][0]["name"] == "span 1" + assert len(envelopes[0].items[0].payload.json["items"]) == 1 + assert envelopes[0].items[0].payload.json["items"][0]["name"] == "span 1" def test_quiet_buckets_flushed_with_busy_neighbors( @@ -344,17 +366,22 @@ def test_quiet_buckets_flushed_with_busy_neighbors( sentry_sdk.traces.new_trace() - with sentry_sdk.traces.start_span(name="span 1") as span1: - trace_id1 = span1.trace_id - - sentry_sdk.traces.new_trace() + with sentry_sdk.new_scope(): + # Keep parent open as its bucket in the batcher would be emptied when it is finished. + parent_a = sentry_sdk.traces.start_span(name="parent a") + with sentry_sdk.traces.start_span(name="span 1", parent_span=parent_a) as span1: + trace_id1 = span1.trace_id - with sentry_sdk.traces.start_span(name="span 2") as span2: - trace_id2 = span2.trace_id + with sentry_sdk.new_scope(): + sentry_sdk.traces.new_trace() + # Keep parent open as its bucket in the batcher would be emptied when it is finished. + parent_b = sentry_sdk.traces.start_span(name="parent b") + with sentry_sdk.traces.start_span(name="span 2", parent_span=parent_b) as span2: + trace_id2 = span2.trace_id - for i in range(3, 10): - with sentry_sdk.traces.start_span(name=f"span {i}"): - pass + for i in range(3, 10): + with sentry_sdk.traces.start_span(name=f"span {i}", parent_span=parent_b): + pass time.sleep(0.3) @@ -432,6 +459,26 @@ def test_transport_format(sentry_init, capture_envelopes): assert value["type"] in ("string", "boolean", "integer", "double", "array") +def test_trace_bucket_flushes_when_segment_ends( + sentry_init, capture_items, monkeypatch +): + """All currently completed spans in a trace are flushed when the segment is finished.""" + 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="segment span"): + with sentry_sdk.traces.start_span(name="child"): + pass + + time.sleep(0.1) + + assert len(items) == 2 + assert items[0].payload["name"] == "child" + assert items[1].payload["name"] == "segment span" + + @pytest.mark.skipif( sys.platform == "win32" or not hasattr(os, "fork") diff --git a/tests/tracing/test_span_streaming.py b/tests/tracing/test_span_streaming.py index 81cbeec7fb..8aef2f7d17 100644 --- a/tests/tracing/test_span_streaming.py +++ b/tests/tracing/test_span_streaming.py @@ -719,14 +719,13 @@ def test_sibling_segments_new_trace(sentry_init, capture_items): spans = [item.payload for item in items] assert len(spans) == 2 - segment1, segment2 = spans - assert segment1["name"] == "segment1" + (segment1,) = (span for span in spans if span["name"] == "segment1") assert segment1["attributes"]["sentry.segment.name"] == "segment1" assert segment1["is_segment"] is True assert "parent_span_id" not in segment1 - assert segment2["name"] == "segment2" + (segment2,) = (span for span in spans if span["name"] == "segment2") assert segment2["attributes"]["sentry.segment.name"] == "segment2" assert segment2["is_segment"] is True assert "parent_span_id" not in segment2