diff --git a/sentry_sdk/integrations/boto3.py b/sentry_sdk/integrations/boto3.py index a7fdd99b21..69deefc7b7 100644 --- a/sentry_sdk/integrations/boto3.py +++ b/sentry_sdk/integrations/boto3.py @@ -67,6 +67,8 @@ def _sentry_request_created( is_span_streaming_enabled = has_span_streaming_enabled(client.options) span: "Union[Span, StreamedSpan]" if is_span_streaming_enabled: + if sentry_sdk.traces.get_current_span() is None: + return span = sentry_sdk.traces.start_span( name=description, attributes={ diff --git a/sentry_sdk/integrations/httpx.py b/sentry_sdk/integrations/httpx.py index a68f20b299..779302a238 100644 --- a/sentry_sdk/integrations/httpx.py +++ b/sentry_sdk/integrations/httpx.py @@ -4,18 +4,15 @@ from sentry_sdk.consts import OP, SPANDATA from sentry_sdk.integrations import DidNotEnable, Integration from sentry_sdk.scope import should_send_default_pii -from sentry_sdk.tracing import BAGGAGE_HEADER_NAME from sentry_sdk.tracing_utils import ( add_http_request_source, - add_sentry_baggage_to_headers, has_span_streaming_enabled, - should_propagate_trace, + propagate_trace_headers, ) from sentry_sdk.utils import ( SENSITIVE_DATA_SUBSTITUTE, capture_internal_exceptions, ensure_integration_enabled, - logger, parse_url, ) @@ -60,6 +57,10 @@ def send(self: "Client", request: "Request", **kwargs: "Any") -> "Response": parsed_url = parse_url(str(request.url), sanitize=False) if is_span_streaming_enabled: + if sentry_sdk.traces.get_current_span() is None: + propagate_trace_headers(client, request) + return real_send(self, request, **kwargs) + with sentry_sdk.traces.start_span( name="%s %s" % ( @@ -81,21 +82,7 @@ def send(self: "Client", request: "Request", **kwargs: "Any") -> "Response": if parsed_url.fragment: attributes["url.fragment"] = parsed_url.fragment - if should_propagate_trace(client, str(request.url)): - for ( - key, - value, - ) in ( - sentry_sdk.get_current_scope().iter_trace_propagation_headers() - ): - logger.debug( - f"[Tracing] Adding `{key}` header {value} to outgoing request to {request.url}." - ) - - if key == BAGGAGE_HEADER_NAME: - add_sentry_baggage_to_headers(request.headers, value) - else: - request.headers[key] = value + propagate_trace_headers(client, request) try: rv = real_send(self, request, **kwargs) @@ -125,21 +112,7 @@ def send(self: "Client", request: "Request", **kwargs: "Any") -> "Response": span.set_data(SPANDATA.HTTP_QUERY, parsed_url.query) span.set_data(SPANDATA.HTTP_FRAGMENT, parsed_url.fragment) - if should_propagate_trace(client, str(request.url)): - for ( - key, - value, - ) in ( - sentry_sdk.get_current_scope().iter_trace_propagation_headers() - ): - logger.debug( - f"[Tracing] Adding `{key}` header {value} to outgoing request to {request.url}." - ) - - if key == BAGGAGE_HEADER_NAME: - add_sentry_baggage_to_headers(request.headers, value) - else: - request.headers[key] = value + propagate_trace_headers(client, request) rv = real_send(self, request, **kwargs) @@ -170,6 +143,10 @@ async def send( parsed_url = parse_url(str(request.url), sanitize=False) if is_span_streaming_enabled: + if sentry_sdk.traces.get_current_span() is None: + propagate_trace_headers(client, request) + return await real_send(self, request, **kwargs) + with sentry_sdk.traces.start_span( name="%s %s" % ( @@ -191,21 +168,7 @@ async def send( if parsed_url.fragment: attributes["url.fragment"] = parsed_url.fragment - if should_propagate_trace(client, str(request.url)): - for ( - key, - value, - ) in ( - sentry_sdk.get_current_scope().iter_trace_propagation_headers() - ): - logger.debug( - f"[Tracing] Adding `{key}` header {value} to outgoing request to {request.url}." - ) - - if key == BAGGAGE_HEADER_NAME: - add_sentry_baggage_to_headers(request.headers, value) - else: - request.headers[key] = value + propagate_trace_headers(client, request) try: rv = await real_send(self, request, **kwargs) @@ -235,20 +198,7 @@ async def send( span.set_data(SPANDATA.HTTP_QUERY, parsed_url.query) span.set_data(SPANDATA.HTTP_FRAGMENT, parsed_url.fragment) - if should_propagate_trace(client, str(request.url)): - for ( - key, - value, - ) in ( - sentry_sdk.get_current_scope().iter_trace_propagation_headers() - ): - logger.debug( - f"[Tracing] Adding `{key}` header {value} to outgoing request to {request.url}." - ) - if key == BAGGAGE_HEADER_NAME: - add_sentry_baggage_to_headers(request.headers, value) - else: - request.headers[key] = value + propagate_trace_headers(client, request) rv = await real_send(self, request, **kwargs) diff --git a/sentry_sdk/integrations/httpx2.py b/sentry_sdk/integrations/httpx2.py index 25062aaa11..1ce3e93524 100644 --- a/sentry_sdk/integrations/httpx2.py +++ b/sentry_sdk/integrations/httpx2.py @@ -4,18 +4,15 @@ from sentry_sdk.consts import OP, SPANDATA from sentry_sdk.integrations import DidNotEnable, Integration from sentry_sdk.scope import should_send_default_pii -from sentry_sdk.tracing import BAGGAGE_HEADER_NAME from sentry_sdk.tracing_utils import ( add_http_request_source, - add_sentry_baggage_to_headers, has_span_streaming_enabled, - should_propagate_trace, + propagate_trace_headers, ) from sentry_sdk.utils import ( SENSITIVE_DATA_SUBSTITUTE, capture_internal_exceptions, ensure_integration_enabled, - logger, parse_url, ) @@ -60,6 +57,11 @@ def send(self: "Client", request: "Request", **kwargs: "Any") -> "Response": parsed_url = parse_url(str(request.url), sanitize=False) if is_span_streaming_enabled: + if sentry_sdk.traces.get_current_span() is None: + propagate_trace_headers(client, request) + + return real_send(self, request, **kwargs) + with sentry_sdk.traces.start_span( name="%s %s" % ( @@ -81,21 +83,7 @@ def send(self: "Client", request: "Request", **kwargs: "Any") -> "Response": if parsed_url.fragment: attributes["url.fragment"] = parsed_url.fragment - if should_propagate_trace(client, str(request.url)): - for ( - key, - value, - ) in ( - sentry_sdk.get_current_scope().iter_trace_propagation_headers() - ): - logger.debug( - f"[Tracing] Adding `{key}` header {value} to outgoing request to {request.url}." - ) - - if key == BAGGAGE_HEADER_NAME: - add_sentry_baggage_to_headers(request.headers, value) - else: - request.headers[key] = value + propagate_trace_headers(client, request) try: rv = real_send(self, request, **kwargs) @@ -125,21 +113,7 @@ def send(self: "Client", request: "Request", **kwargs: "Any") -> "Response": span.set_data(SPANDATA.HTTP_QUERY, parsed_url.query) span.set_data(SPANDATA.HTTP_FRAGMENT, parsed_url.fragment) - if should_propagate_trace(client, str(request.url)): - for ( - key, - value, - ) in ( - sentry_sdk.get_current_scope().iter_trace_propagation_headers() - ): - logger.debug( - f"[Tracing] Adding `{key}` header {value} to outgoing request to {request.url}." - ) - - if key == BAGGAGE_HEADER_NAME: - add_sentry_baggage_to_headers(request.headers, value) - else: - request.headers[key] = value + propagate_trace_headers(client, request) rv = real_send(self, request, **kwargs) @@ -170,6 +144,11 @@ async def send( parsed_url = parse_url(str(request.url), sanitize=False) if is_span_streaming_enabled: + if sentry_sdk.traces.get_current_span() is None: + propagate_trace_headers(client, request) + + return await real_send(self, request, **kwargs) + with sentry_sdk.traces.start_span( name="%s %s" % ( @@ -191,21 +170,7 @@ async def send( if parsed_url.fragment: attributes["url.fragment"] = parsed_url.fragment - if should_propagate_trace(client, str(request.url)): - for ( - key, - value, - ) in ( - sentry_sdk.get_current_scope().iter_trace_propagation_headers() - ): - logger.debug( - f"[Tracing] Adding `{key}` header {value} to outgoing request to {request.url}." - ) - - if key == BAGGAGE_HEADER_NAME: - add_sentry_baggage_to_headers(request.headers, value) - else: - request.headers[key] = value + propagate_trace_headers(client, request) try: rv = await real_send(self, request, **kwargs) @@ -235,20 +200,7 @@ async def send( span.set_data(SPANDATA.HTTP_QUERY, parsed_url.query) span.set_data(SPANDATA.HTTP_FRAGMENT, parsed_url.fragment) - if should_propagate_trace(client, str(request.url)): - for ( - key, - value, - ) in ( - sentry_sdk.get_current_scope().iter_trace_propagation_headers() - ): - logger.debug( - f"[Tracing] Adding `{key}` header {value} to outgoing request to {request.url}." - ) - if key == BAGGAGE_HEADER_NAME: - add_sentry_baggage_to_headers(request.headers, value) - else: - request.headers[key] = value + propagate_trace_headers(client, request) rv = await real_send(self, request, **kwargs) diff --git a/sentry_sdk/integrations/pyreqwest.py b/sentry_sdk/integrations/pyreqwest.py index aae68c4c10..d25d03f470 100644 --- a/sentry_sdk/integrations/pyreqwest.py +++ b/sentry_sdk/integrations/pyreqwest.py @@ -12,6 +12,7 @@ add_http_request_source, add_sentry_baggage_to_headers, has_span_streaming_enabled, + propagate_trace_headers, should_propagate_trace, ) from sentry_sdk.utils import ( @@ -88,6 +89,11 @@ def _sentry_pyreqwest_span(request: "Request") -> "Generator[Any, None, None]": span_streaming = has_span_streaming_enabled(sentry_sdk.get_client().options) if span_streaming: + if sentry_sdk.traces.get_current_span() is None: + propagate_trace_headers(client=sentry_sdk.get_client(), request=request) + yield None + return + with sentry_sdk.traces.start_span( name=f"{request.method} {parsed_url.url if parsed_url else SENSITIVE_DATA_SUBSTITUTE}", attributes={ @@ -101,26 +107,13 @@ def _sentry_pyreqwest_span(request: "Request") -> "Generator[Any, None, None]": span.set_attribute(SPANDATA.URL_QUERY, parsed_url.query) span.set_attribute(SPANDATA.URL_FRAGMENT, parsed_url.fragment) - if should_propagate_trace(sentry_sdk.get_client(), str(request.url)): - for ( - key, - value, - ) in sentry_sdk.get_current_scope().iter_trace_propagation_headers(): - logger.debug( - "[Tracing] Adding `{key}` header {value} to outgoing request to {url}.".format( - key=key, value=value, url=request.url - ) - ) - - if key == BAGGAGE_HEADER_NAME: - add_sentry_baggage_to_headers(request.headers, value) - else: - request.headers[key] = value + propagate_trace_headers(client=sentry_sdk.get_client(), request=request) yield span - with capture_internal_exceptions(): - add_http_request_source(span) + if span is not None: + with capture_internal_exceptions(): + add_http_request_source(span) return @@ -171,7 +164,7 @@ async def sentry_async_middleware( SPANDATA.HTTP_STATUS_CODE, response.status, ) - else: + elif span is not None: span.set_http_status(response.status) return response @@ -191,7 +184,7 @@ def sentry_sync_middleware( SPANDATA.HTTP_STATUS_CODE, response.status, ) - else: + elif span is not None: span.set_http_status(response.status) return response diff --git a/sentry_sdk/integrations/stdlib.py b/sentry_sdk/integrations/stdlib.py index 82f30f2dda..4de3819a77 100644 --- a/sentry_sdk/integrations/stdlib.py +++ b/sentry_sdk/integrations/stdlib.py @@ -112,25 +112,30 @@ def putrequest( parsed_url = parse_url(real_url, sanitize=False) span_streaming = has_span_streaming_enabled(client.options) - span: "Union[Span, StreamedSpan]" + span: "Union[Span, StreamedSpan, None]" if span_streaming: - span = sentry_sdk.traces.start_span( - name="%s %s" - % (method, parsed_url.url if parsed_url else SENSITIVE_DATA_SUBSTITUTE), - attributes={ - "sentry.origin": "auto.http.stdlib.httplib", - "sentry.op": OP.HTTP_CLIENT, - SPANDATA.HTTP_REQUEST_METHOD: method, - }, - ) - - if parsed_url is not None and should_send_default_pii(): - span.set_attribute(SPANDATA.URL_FRAGMENT, parsed_url.fragment) - span.set_attribute(SPANDATA.URL_FULL, parsed_url.url) - span.set_attribute(SPANDATA.URL_QUERY, parsed_url.query) + if sentry_sdk.traces.get_current_span() is None: + span = None + else: + span = sentry_sdk.traces.start_span( + name="%s %s" + % ( + method, + parsed_url.url if parsed_url else SENSITIVE_DATA_SUBSTITUTE, + ), + attributes={ + "sentry.origin": "auto.http.stdlib.httplib", + "sentry.op": OP.HTTP_CLIENT, + SPANDATA.HTTP_REQUEST_METHOD: method, + }, + ) - set_on_span = span.set_attribute + if parsed_url is not None and should_send_default_pii(): + span.set_attribute(SPANDATA.URL_FRAGMENT, parsed_url.fragment) + span.set_attribute(SPANDATA.URL_FULL, parsed_url.url) + span.set_attribute(SPANDATA.URL_QUERY, parsed_url.query) + set_on_span = span.set_attribute else: span = sentry_sdk.start_span( op=OP.HTTP_CLIENT, @@ -148,7 +153,7 @@ def putrequest( set_on_span = span.set_data # for proxies, these point to the proxy host/port - if tunnel_host: + if span and tunnel_host: set_on_span(SPANDATA.NETWORK_PEER_ADDRESS, self.host) set_on_span(SPANDATA.NETWORK_PEER_PORT, self.port) @@ -303,6 +308,9 @@ def sentry_patched_popen_init( span_streaming = has_span_streaming_enabled(sentry_sdk.get_client().options) span: "Union[Span, StreamedSpan]" if span_streaming: + if sentry_sdk.traces.get_current_span() is None: + return old_popen_init(self, *a, **kw) + span = sentry_sdk.traces.start_span( name=description, attributes={ @@ -353,6 +361,8 @@ def sentry_patched_popen_wait( ) -> "Any": span_streaming = has_span_streaming_enabled(sentry_sdk.get_client().options) if span_streaming: + if sentry_sdk.traces.get_current_span() is None: + return old_popen_wait(self, *a, **kw) with sentry_sdk.traces.start_span( name=OP.SUBPROCESS_WAIT, attributes={ @@ -380,6 +390,8 @@ def sentry_patched_popen_communicate( ) -> "Any": span_streaming = has_span_streaming_enabled(sentry_sdk.get_client().options) if span_streaming: + if sentry_sdk.traces.get_current_span() is None: + return old_popen_communicate(self, *a, **kw) with sentry_sdk.traces.start_span( name=OP.SUBPROCESS_COMMUNICATE, attributes={ diff --git a/sentry_sdk/tracing_utils.py b/sentry_sdk/tracing_utils.py index 8cb1cbe3ae..8cf11d01f7 100644 --- a/sentry_sdk/tracing_utils.py +++ b/sentry_sdk/tracing_utils.py @@ -973,6 +973,37 @@ def should_propagate_trace(client: "sentry_sdk.client.BaseClient", url: str) -> return match_regex_list(url, trace_propagation_targets, substring_matching=True) +def propagate_trace_headers( + client: "sentry_sdk.client.BaseClient", request: "Any" +) -> None: + """ + Attach Sentry trace propagation headers (``sentry-trace``/``baggage``) from the + current scope's propagation context to an outgoing request, if the request's + URL matches the configured ``trace_propagation_targets``. + + ``request`` is expected to expose ``url`` and a mutable ``headers`` mapping + (e.g. an ``httpx``/``httpx2`` ``Request``). + """ + if not hasattr(request, "url") or not hasattr(request, "headers"): + logger.warning( + "Unable to propagate trace headers in request - missing url or headers attributes" + ) + return + + if not should_propagate_trace(client, str(request.url)): + return + + for key, value in sentry_sdk.get_current_scope().iter_trace_propagation_headers(): + logger.debug( + f"[Tracing] Adding `{key}` header {value} to outgoing request to {request.url}." + ) + + if key == BAGGAGE_HEADER_NAME: + add_sentry_baggage_to_headers(request.headers, value) + else: + request.headers[key] = value + + def normalize_incoming_data(incoming_data: "Dict[str, Any]") -> "Dict[str, Any]": """ Normalizes incoming data so the keys are all lowercase with dashes instead of underscores and stripped from known prefixes. diff --git a/tests/integrations/arq/test_arq.py b/tests/integrations/arq/test_arq.py index 013ea83167..d962bf61c7 100644 --- a/tests/integrations/arq/test_arq.py +++ b/tests/integrations/arq/test_arq.py @@ -248,19 +248,19 @@ async def retry_job(ctx): # The retry re-enqueue happens without an active span, so no producer # (queue.submit.arq) span is created for it; only the consumer segment # is emitted. The consumer segment is preceded by the redis spans for - # the re-enqueue, so it lands at index 3. - assert spans[3]["attributes"]["sentry.op"] == "queue.task.arq" - assert spans[3]["status"] == "ok" - assert spans[3]["name"] == "retry_job" + # the re-enqueue, so it lands at index 2. + assert spans[2]["attributes"]["sentry.op"] == "queue.task.arq" + assert spans[2]["status"] == "ok" + assert spans[2]["name"] == "retry_job" await worker.run_job(job.job_id, timestamp_ms()) sentry_sdk.flush() spans = [item.payload for item in items] - assert spans[6]["attributes"]["sentry.op"] == "queue.task.arq" - assert spans[6]["status"] == "ok" - assert spans[6]["name"] == "retry_job" + assert spans[5]["attributes"]["sentry.op"] == "queue.task.arq" + assert spans[5]["status"] == "ok" + assert spans[5]["name"] == "retry_job" else: events = capture_events() @@ -589,7 +589,7 @@ async def job(ctx): pool, worker = init_fixture_method(span_streaming, [job]) if span_streaming: - job = await pool.enqueue_job("retry_job") + job = await pool.enqueue_job("job") items = capture_items("span") @@ -600,13 +600,13 @@ async def job(ctx): # No producer (queue.submit.arq) span is created for the re-enqueue # triggered by the retry, since it happens without an active span, so - # the consumer segment lands at index 3. - assert spans[3]["attributes"]["sentry.op"] == "queue.task.arq" - assert spans[3]["attributes"]["sentry.origin"] == "auto.queue.arq" - assert spans[2]["attributes"]["sentry.origin"] == "auto.db.redis" + # the consumer segment lands at index 2. + assert spans[2]["attributes"]["sentry.op"] == "queue.task.arq" + assert spans[2]["attributes"]["sentry.origin"] == "auto.queue.arq" assert spans[1]["attributes"]["sentry.origin"] == "auto.db.redis" + assert spans[0]["attributes"]["sentry.origin"] == "auto.db.redis" else: - job = await pool.enqueue_job("retry_job") + job = await pool.enqueue_job("job") events = capture_events() diff --git a/tests/integrations/httpx/test_httpx.py b/tests/integrations/httpx/test_httpx.py index 4ca18c8edc..c0a0ea049c 100644 --- a/tests/integrations/httpx/test_httpx.py +++ b/tests/integrations/httpx/test_httpx.py @@ -739,10 +739,13 @@ def test_outgoing_trace_headers_span_streaming( items = capture_items("span") - if asyncio.iscoroutinefunction(httpx_client.get): - response = asyncio.get_event_loop().run_until_complete(httpx_client.get(url)) - else: - response = httpx_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx_client.get): + response = asyncio.get_event_loop().run_until_complete( + httpx_client.get(url) + ) + else: + response = httpx_client.get(url) sentry_sdk.flush() @@ -781,12 +784,13 @@ def test_outgoing_trace_headers_append_to_baggage_span_streaming( items = capture_items("span") with mock.patch("sentry_sdk.tracing_utils.Random.randrange", return_value=500000): - if asyncio.iscoroutinefunction(httpx_client.get): - response = asyncio.get_event_loop().run_until_complete( - httpx_client.get(url, headers={"baGGage": "custom=data"}) - ) - else: - response = httpx_client.get(url, headers={"baGGage": "custom=data"}) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx_client.get): + response = asyncio.get_event_loop().run_until_complete( + httpx_client.get(url, headers={"baGGage": "custom=data"}) + ) + else: + response = httpx_client.get(url, headers={"baGGage": "custom=data"}) sentry_sdk.flush() @@ -799,6 +803,87 @@ def test_outgoing_trace_headers_append_to_baggage_span_streaming( assert "sentry-sampled=true" in baggage +def test_outgoing_trace_headers_span_streaming_no_current_span(sentry_init, httpx_mock): + """ + Even when there is no active span, trace propagation headers should still + be attached to outgoing requests when span streaming is enabled. + + This is deliberately different from the legacy (transaction-based) approach, + which does not propagate outside of a transaction (see + ``test_do_not_propagate_outside_transaction``). The streamed approach + propagates from the current scope's propagation context regardless of + whether a span is active. + """ + httpx_mock.add_response() + + sentry_init( + traces_sample_rate=1.0, + trace_propagation_targets=[MATCH_ALL], + integrations=[HttpxIntegration()], + _experiments={"trace_lifecycle": "stream"}, + ) + + url = "http://example.com/" + + httpx_client = httpx.Client() + + # No start_span / start_transaction -> get_current_span() is None + assert sentry_sdk.traces.get_current_span() is None + + response = httpx_client.get(url) + + assert response.status_code == 200 + + # Trace is still propagated from the scope's propagation context + request_headers = httpx_mock.get_request().headers + assert "sentry-trace" in request_headers + assert "baggage" in request_headers + + # The propagated headers describe a single, coherent trace: the trace_id in + # sentry-trace matches the one carried in baggage. + trace_id = request_headers["sentry-trace"].split("-")[0] + assert f"sentry-trace_id={trace_id}" in request_headers["baggage"] + + +def test_outgoing_trace_headers_span_streaming_no_current_span_async( + sentry_init, httpx_mock +): + """ + The async client must match the sync client: trace propagation headers are + attached to outgoing requests even when there is no active span and span + streaming is enabled. + """ + httpx_mock.add_response() + + sentry_init( + traces_sample_rate=1.0, + trace_propagation_targets=[MATCH_ALL], + integrations=[HttpxIntegration()], + _experiments={"trace_lifecycle": "stream"}, + ) + + url = "http://example.com/" + + httpx_client = httpx.AsyncClient() + + # No start_span / start_transaction -> get_current_span() is None + assert sentry_sdk.traces.get_current_span() is None + + response = asyncio.get_event_loop().run_until_complete(httpx_client.get(url)) + + assert response.status_code == 200 + + # Trace is still propagated from the scope's propagation context + request_headers = httpx_mock.get_request().headers + assert "sentry-trace" in request_headers + assert "baggage" in request_headers + + # The propagated headers describe a single, coherent trace: the trace_id in + # sentry-trace matches the one carried in baggage. + trace_id = request_headers["sentry-trace"].split("-")[0] + assert f"sentry-trace_id={trace_id}" in request_headers["baggage"] + + @pytest.mark.parametrize( "httpx_client", (httpx.Client(), httpx.AsyncClient()), @@ -820,10 +905,11 @@ def test_request_source_disabled_span_streaming( url = "http://example.com/" - if asyncio.iscoroutinefunction(httpx_client.get): - asyncio.get_event_loop().run_until_complete(httpx_client.get(url)) - else: - httpx_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx_client.get): + asyncio.get_event_loop().run_until_complete(httpx_client.get(url)) + else: + httpx_client.get(url) sentry_sdk.flush() @@ -864,10 +950,11 @@ def test_request_source_enabled_span_streaming( url = "http://example.com/" - if asyncio.iscoroutinefunction(httpx_client.get): - asyncio.get_event_loop().run_until_complete(httpx_client.get(url)) - else: - httpx_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx_client.get): + asyncio.get_event_loop().run_until_complete(httpx_client.get(url)) + else: + httpx_client.get(url) sentry_sdk.flush() @@ -900,10 +987,11 @@ def test_request_source_span_streaming( url = "http://example.com/" - if asyncio.iscoroutinefunction(httpx_client.get): - asyncio.get_event_loop().run_until_complete(httpx_client.get(url)) - else: - httpx_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx_client.get): + asyncio.get_event_loop().run_until_complete(httpx_client.get(url)) + else: + httpx_client.get(url) sentry_sdk.flush() @@ -957,16 +1045,17 @@ def test_request_source_with_module_in_search_path_span_streaming( url = "http://example.com/" - if asyncio.iscoroutinefunction(httpx_client.get): - from httpx_helpers.helpers import async_get_request_with_client + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx_client.get): + from httpx_helpers.helpers import async_get_request_with_client - asyncio.get_event_loop().run_until_complete( - async_get_request_with_client(httpx_client, url) - ) - else: - from httpx_helpers.helpers import get_request_with_client + asyncio.get_event_loop().run_until_complete( + async_get_request_with_client(httpx_client, url) + ) + else: + from httpx_helpers.helpers import get_request_with_client - get_request_with_client(httpx_client, url) + get_request_with_client(httpx_client, url) sentry_sdk.flush() @@ -1018,10 +1107,11 @@ def test_no_request_source_if_duration_too_short_span_streaming( url = "http://example.com/" - if asyncio.iscoroutinefunction(httpx_client.get): - asyncio.get_event_loop().run_until_complete(httpx_client.get(url)) - else: - httpx_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx_client.get): + asyncio.get_event_loop().run_until_complete(httpx_client.get(url)) + else: + httpx_client.get(url) sentry_sdk.flush() @@ -1055,10 +1145,11 @@ def test_request_source_if_duration_over_threshold_span_streaming( url = "http://example.com/" - if asyncio.iscoroutinefunction(httpx_client.get): - asyncio.get_event_loop().run_until_complete(httpx_client.get(url)) - else: - httpx_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx_client.get): + asyncio.get_event_loop().run_until_complete(httpx_client.get(url)) + else: + httpx_client.get(url) sentry_sdk.flush() @@ -1107,10 +1198,11 @@ def test_span_origin_span_streaming( url = "http://example.com/" - if asyncio.iscoroutinefunction(httpx_client.get): - asyncio.get_event_loop().run_until_complete(httpx_client.get(url)) - else: - httpx_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx_client.get): + asyncio.get_event_loop().run_until_complete(httpx_client.get(url)) + else: + httpx_client.get(url) sentry_sdk.flush() @@ -1140,10 +1232,11 @@ def test_http_url_attributes_span_streaming( url = "http://example.com/?foo=bar#frag" - if asyncio.iscoroutinefunction(httpx_client.get): - asyncio.get_event_loop().run_until_complete(httpx_client.get(url)) - else: - httpx_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx_client.get): + asyncio.get_event_loop().run_until_complete(httpx_client.get(url)) + else: + httpx_client.get(url) sentry_sdk.flush() @@ -1183,10 +1276,11 @@ def test_http_url_attributes_no_query_or_fragment_span_streaming( url = "http://example.com/" - if asyncio.iscoroutinefunction(httpx_client.get): - asyncio.get_event_loop().run_until_complete(httpx_client.get(url)) - else: - httpx_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx_client.get): + asyncio.get_event_loop().run_until_complete(httpx_client.get(url)) + else: + httpx_client.get(url) sentry_sdk.flush() diff --git a/tests/integrations/httpx2/test_httpx2.py b/tests/integrations/httpx2/test_httpx2.py index 15d50513ae..d66c62849b 100644 --- a/tests/integrations/httpx2/test_httpx2.py +++ b/tests/integrations/httpx2/test_httpx2.py @@ -733,10 +733,11 @@ def test_outgoing_trace_headers_span_streaming( items = capture_items("span") - if asyncio.iscoroutinefunction(httpx2_client.get): - response = asyncio.run(httpx2_client.get(url)) - else: - response = httpx2_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx2_client.get): + response = asyncio.run(httpx2_client.get(url)) + else: + response = httpx2_client.get(url) sentry_sdk.flush() @@ -775,12 +776,13 @@ def test_outgoing_trace_headers_append_to_baggage_span_streaming( items = capture_items("span") with mock.patch("sentry_sdk.tracing_utils.Random.randrange", return_value=500000): - if asyncio.iscoroutinefunction(httpx2_client.get): - response = asyncio.run( - httpx2_client.get(url, headers={"baGGage": "custom=data"}) - ) - else: - response = httpx2_client.get(url, headers={"baGGage": "custom=data"}) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx2_client.get): + response = asyncio.run( + httpx2_client.get(url, headers={"baGGage": "custom=data"}) + ) + else: + response = httpx2_client.get(url, headers={"baGGage": "custom=data"}) sentry_sdk.flush() @@ -793,6 +795,89 @@ def test_outgoing_trace_headers_append_to_baggage_span_streaming( assert "sentry-sampled=true" in baggage +def test_outgoing_trace_headers_span_streaming_no_current_span( + sentry_init, httpx2_mock +): + """ + Even when there is no active span, trace propagation headers should still + be attached to outgoing requests when span streaming is enabled. + + This is deliberately different from the legacy (transaction-based) approach, + which does not propagate outside of a transaction (see + ``test_do_not_propagate_outside_transaction``). The streamed approach + propagates from the current scope's propagation context regardless of + whether a span is active. + """ + httpx2_mock.add_response() + + sentry_init( + traces_sample_rate=1.0, + trace_propagation_targets=[MATCH_ALL], + integrations=[Httpx2Integration()], + _experiments={"trace_lifecycle": "stream"}, + ) + + url = "http://example.com/" + + httpx2_client = httpx2.Client() + + # No start_span / start_transaction -> get_current_span() is None + assert sentry_sdk.traces.get_current_span() is None + + response = httpx2_client.get(url) + + assert response.status_code == 200 + + # Trace is still propagated from the scope's propagation context + request_headers = httpx2_mock.get_request().headers + assert "sentry-trace" in request_headers + assert "baggage" in request_headers + + # The propagated headers describe a single, coherent trace: the trace_id in + # sentry-trace matches the one carried in baggage. + trace_id = request_headers["sentry-trace"].split("-")[0] + assert f"sentry-trace_id={trace_id}" in request_headers["baggage"] + + +def test_outgoing_trace_headers_span_streaming_no_current_span_async( + sentry_init, httpx2_mock +): + """ + The async client must match the sync client: trace propagation headers are + attached to outgoing requests even when there is no active span and span + streaming is enabled. + """ + httpx2_mock.add_response() + + sentry_init( + traces_sample_rate=1.0, + trace_propagation_targets=[MATCH_ALL], + integrations=[Httpx2Integration()], + _experiments={"trace_lifecycle": "stream"}, + ) + + url = "http://example.com/" + + httpx2_client = httpx2.AsyncClient() + + # No start_span / start_transaction -> get_current_span() is None + assert sentry_sdk.traces.get_current_span() is None + + response = asyncio.run(httpx2_client.get(url)) + + assert response.status_code == 200 + + # Trace is still propagated from the scope's propagation context + request_headers = httpx2_mock.get_request().headers + assert "sentry-trace" in request_headers + assert "baggage" in request_headers + + # The propagated headers describe a single, coherent trace: the trace_id in + # sentry-trace matches the one carried in baggage. + trace_id = request_headers["sentry-trace"].split("-")[0] + assert f"sentry-trace_id={trace_id}" in request_headers["baggage"] + + @pytest.mark.parametrize( "httpx2_client", (httpx2.Client(), httpx2.AsyncClient()), @@ -814,10 +899,11 @@ def test_request_source_disabled_span_streaming( url = "http://example.com/" - if asyncio.iscoroutinefunction(httpx2_client.get): - asyncio.run(httpx2_client.get(url)) - else: - httpx2_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx2_client.get): + asyncio.run(httpx2_client.get(url)) + else: + httpx2_client.get(url) sentry_sdk.flush() @@ -858,10 +944,11 @@ def test_request_source_enabled_span_streaming( url = "http://example.com/" - if asyncio.iscoroutinefunction(httpx2_client.get): - asyncio.run(httpx2_client.get(url)) - else: - httpx2_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx2_client.get): + asyncio.run(httpx2_client.get(url)) + else: + httpx2_client.get(url) sentry_sdk.flush() @@ -894,10 +981,11 @@ def test_request_source_span_streaming( url = "http://example.com/" - if asyncio.iscoroutinefunction(httpx2_client.get): - asyncio.run(httpx2_client.get(url)) - else: - httpx2_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx2_client.get): + asyncio.run(httpx2_client.get(url)) + else: + httpx2_client.get(url) sentry_sdk.flush() @@ -951,14 +1039,15 @@ def test_request_source_with_module_in_search_path_span_streaming( url = "http://example.com/" - if asyncio.iscoroutinefunction(httpx2_client.get): - from httpx2_helpers.helpers import async_get_request_with_client + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx2_client.get): + from httpx2_helpers.helpers import async_get_request_with_client - asyncio.run(async_get_request_with_client(httpx2_client, url)) - else: - from httpx2_helpers.helpers import get_request_with_client + asyncio.run(async_get_request_with_client(httpx2_client, url)) + else: + from httpx2_helpers.helpers import get_request_with_client - get_request_with_client(httpx2_client, url) + get_request_with_client(httpx2_client, url) sentry_sdk.flush() @@ -1010,10 +1099,11 @@ def test_no_request_source_if_duration_too_short_span_streaming( url = "http://example.com/" - if asyncio.iscoroutinefunction(httpx2_client.get): - asyncio.run(httpx2_client.get(url)) - else: - httpx2_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx2_client.get): + asyncio.run(httpx2_client.get(url)) + else: + httpx2_client.get(url) sentry_sdk.flush() @@ -1047,10 +1137,11 @@ def test_request_source_if_duration_over_threshold_span_streaming( url = "http://example.com/" - if asyncio.iscoroutinefunction(httpx2_client.get): - asyncio.run(httpx2_client.get(url)) - else: - httpx2_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx2_client.get): + asyncio.run(httpx2_client.get(url)) + else: + httpx2_client.get(url) sentry_sdk.flush() @@ -1099,10 +1190,11 @@ def test_span_origin_span_streaming( url = "http://example.com/" - if asyncio.iscoroutinefunction(httpx2_client.get): - asyncio.run(httpx2_client.get(url)) - else: - httpx2_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx2_client.get): + asyncio.run(httpx2_client.get(url)) + else: + httpx2_client.get(url) sentry_sdk.flush() @@ -1131,10 +1223,11 @@ def test_http_url_attributes_span_streaming( url = "http://example.com/?foo=bar#frag" - if asyncio.iscoroutinefunction(httpx2_client.get): - asyncio.run(httpx2_client.get(url)) - else: - httpx2_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx2_client.get): + asyncio.run(httpx2_client.get(url)) + else: + httpx2_client.get(url) sentry_sdk.flush() @@ -1167,10 +1260,11 @@ def test_http_url_attributes_no_query_or_fragment_span_streaming( url = "http://example.com/" - if asyncio.iscoroutinefunction(httpx2_client.get): - asyncio.run(httpx2_client.get(url)) - else: - httpx2_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx2_client.get): + asyncio.run(httpx2_client.get(url)) + else: + httpx2_client.get(url) sentry_sdk.flush() @@ -1202,10 +1296,11 @@ def test_http_url_attributes_pii_disabled_span_streaming( url = "http://example.com/?foo=bar#frag" - if asyncio.iscoroutinefunction(httpx2_client.get): - asyncio.run(httpx2_client.get(url)) - else: - httpx2_client.get(url) + with sentry_sdk.traces.start_span(name="test"): + if asyncio.iscoroutinefunction(httpx2_client.get): + asyncio.run(httpx2_client.get(url)) + else: + httpx2_client.get(url) sentry_sdk.flush() diff --git a/tests/integrations/stdlib/test_httplib.py b/tests/integrations/stdlib/test_httplib.py index 64e059fc4c..f780109ab6 100644 --- a/tests/integrations/stdlib/test_httplib.py +++ b/tests/integrations/stdlib/test_httplib.py @@ -460,6 +460,74 @@ def getresponse(self, *args, **kwargs): assert request_headers["baggage"] == expected_outgoing_baggage +def test_outgoing_trace_headers_span_streaming_no_current_span(sentry_init): + """ + With span streaming enabled and no active span, trace propagation headers + should still be attached to outgoing requests, propagated from the scope's + propagation context. + """ + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream"}, + ) + + already_patched_getresponse = HTTPSConnection.getresponse + request_headers = {} + + class HTTPSConnectionRecordingRequestHeaders(HTTPSConnection): + def send(self, *args, **kwargs) -> None: + request_str = args[0] + for line in request_str.decode("utf-8").split("\r\n")[1:]: + if line: + key, val = line.split(": ") + request_headers[key] = val + + server_sock, client_sock = socket.socketpair() + server_sock.sendall(b"HTTP/1.1 200 OK\r\n\r\n") + server_sock.close() + self.sock = client_sock + + def getresponse(self, *args, **kwargs): + return already_patched_getresponse(self, *args, **kwargs) + + headers = { + "sentry-trace": "771a43a4192642f0b136d5159a501700-1234567890abcdef-1", + "baggage": ( + "sentry-trace_id=771a43a4192642f0b136d5159a501700," + "sentry-public_key=49d0f7386ad645858ae85020e393bef3," + "sentry-sample_rate=0.01337," + "sentry-sample_rand=0.000005," + "sentry-sampled=true" + ), + } + + # Seed the scope's propagation context, but do NOT start a span. + sentry_sdk.traces.continue_trace(headers) + assert sentry_sdk.traces.get_current_span() is None + + connection = HTTPSConnectionRecordingRequestHeaders("localhost", port=PORT) + connection.request("GET", "/top-chasers") + connection.getresponse() + + # Trace is still propagated, carrying the trace_id from the propagation + # context. The span id segment is generated for the propagation context + # (there is no active span), so we assert the stable trace_id prefix. + assert request_headers["sentry-trace"].startswith( + "771a43a4192642f0b136d5159a501700-" + ) + + # The outgoing baggage is fully deterministic: it is the frozen incoming + # baggage from the propagation context. + expected_outgoing_baggage = ( + "sentry-trace_id=771a43a4192642f0b136d5159a501700," + "sentry-public_key=49d0f7386ad645858ae85020e393bef3," + "sentry-sample_rate=0.01337," + "sentry-sample_rand=0.000005," + "sentry-sampled=true" + ) + assert request_headers["baggage"] == expected_outgoing_baggage + + @pytest.mark.parametrize( "trace_propagation_targets,host,path,trace_propagated", [ diff --git a/tests/integrations/strawberry/test_strawberry.py b/tests/integrations/strawberry/test_strawberry.py index 16e15142ff..99802a3a86 100644 --- a/tests/integrations/strawberry/test_strawberry.py +++ b/tests/integrations/strawberry/test_strawberry.py @@ -331,14 +331,8 @@ def test_capture_transaction_on_error( assert len(error_events) == 1 - if async_execution: - # When FastAPI is run, there's an extra span from the httpx client - # so we need to account for that - assert len(spans) == 6 - parse_span, validate_span, resolve_span, query_span, segment, _ = spans - else: - assert len(spans) == 5 - parse_span, validate_span, resolve_span, query_span, segment = spans + assert len(spans) == 5 + parse_span, validate_span, resolve_span, query_span, segment = spans assert segment["is_segment"] is True assert segment["name"] == "ErrorQuery" @@ -473,12 +467,8 @@ def test_capture_transaction_on_success( sentry_sdk.flush() spans = [i.payload for i in items] - if async_execution: - assert len(spans) == 6 - parse_span, validate_span, resolve_span, query_span, segment, _ = spans - else: - assert len(spans) == 5 - parse_span, validate_span, resolve_span, query_span, segment = spans + assert len(spans) == 5 + parse_span, validate_span, resolve_span, query_span, segment = spans assert segment["is_segment"] is True assert segment["name"] == "GreetingQuery" @@ -613,12 +603,8 @@ def test_transaction_no_operation_name( sentry_sdk.flush() spans = [i.payload for i in items] - if async_execution: - assert len(spans) == 6 - parse_span, validate_span, resolve_span, query_span, segment, _ = spans - else: - assert len(spans) == 5 - parse_span, validate_span, resolve_span, query_span, segment = spans + assert len(spans) == 5 + parse_span, validate_span, resolve_span, query_span, segment = spans assert segment["is_segment"] is True if async_execution: @@ -758,12 +744,8 @@ def test_transaction_mutation( sentry_sdk.flush() spans = [i.payload for i in items] - if async_execution: - assert len(spans) == 6 - parse_span, validate_span, resolve_span, mutation_span, segment, _ = spans - else: - assert len(spans) == 5 - parse_span, validate_span, resolve_span, mutation_span, segment = spans + assert len(spans) == 5 + parse_span, validate_span, resolve_span, mutation_span, segment = spans assert segment["is_segment"] is True assert segment["name"] == "Change" @@ -924,12 +906,8 @@ def test_span_origin( sentry_sdk.flush() spans = [i.payload for i in items] - if async_execution: - assert len(spans) == 6 - parse_span, validate_span, resolve_span, mutation_span, segment, _ = spans - else: - assert len(spans) == 5 - parse_span, validate_span, resolve_span, mutation_span, segment = spans + assert len(spans) == 5 + parse_span, validate_span, resolve_span, mutation_span, segment = spans assert segment["is_segment"] is True if is_flask: @@ -995,12 +973,8 @@ def test_span_origin2( sentry_sdk.flush() spans = [i.payload for i in items] - if async_execution: - assert len(spans) == 6 - parse_span, validate_span, resolve_span, query_span, segment, _ = spans - else: - assert len(spans) == 5 - parse_span, validate_span, resolve_span, query_span, segment = spans + assert len(spans) == 5 + parse_span, validate_span, resolve_span, query_span, segment = spans assert segment["is_segment"] is True if is_flask: @@ -1066,14 +1040,8 @@ def test_span_origin3( sentry_sdk.flush() spans = [i.payload for i in items] - if async_execution: - assert len(spans) == 6 - parse_span, validate_span, resolve_span, subscription_span, segment, _ = ( - spans - ) - else: - assert len(spans) == 5 - parse_span, validate_span, resolve_span, subscription_span, segment = spans + assert len(spans) == 5 + parse_span, validate_span, resolve_span, subscription_span, segment = spans assert segment["is_segment"] is True if is_flask: