feat(client): Use span streaming exclusively

clients/python/pyproject.toml

@@ -5,7 +5,7 @@ description = "Taskbroker python client and worker runtime"
 readme = "README.md"
 dependencies = [
     "sentry-arroyo>=2.41.1",
-    "sentry-sdk[http2]>=2.66.1",
+    "sentry-sdk[http2]>=2.73.0",
     "sentry-protos>=0.26.1",
     "confluent_kafka>=2.3.0",
     "cronsim>=2.6",

clients/python/src/taskbroker_client/sdk.py

@@ -4,82 +4,53 @@
 import sentry_sdk
 from sentry_sdk.scope import Scope
 from sentry_sdk.traces import StreamedSpan
-from sentry_sdk.tracing import NoOpSpan, Span, Transaction
-from sentry_sdk.tracing_utils import has_span_streaming_enabled
 
 
-def start_transaction(
+def start_segment(
     name: str,
     op: str,
     origin: str,
     attributes: dict[str, Any],
     headers: dict[str, Any],
     sampling_context: dict[str, Any],
-) -> Transaction | NoOpSpan | StreamedSpan | ContextManager[Any]:
-    """Start a transaction, or a span if span streaming is enabled."""
-    span = None
+) -> StreamedSpan | ContextManager[Any]:
+    """Continue the trace from `headers` and start a segment span."""
     try:
-        is_span_streaming = has_span_streaming_enabled(sentry_sdk.get_client().options)
-        if is_span_streaming:
-            sentry_sdk.traces.continue_trace(headers)
-            Scope.set_custom_sampling_context(sampling_context)
+        sentry_sdk.traces.continue_trace(headers)
+        Scope.set_custom_sampling_context(sampling_context)
 
-            return sentry_sdk.traces.start_span(
-                name=name,
-                attributes={
-                    "sentry.op": op,
-                    **attributes,
-                },
-            )
-
-        transaction = sentry_sdk.continue_trace(
-            environ_or_headers=headers,
-            op=op,
+        return sentry_sdk.traces.start_span(
             name=name,
-            origin=origin,
+            attributes={
+                "sentry.op": op,
+                "sentry.origin": origin,
+                **attributes,
+            },
         )
-
-        span = sentry_sdk.start_transaction(transaction, custom_sampling_context=sampling_context)
-        for key, value in attributes.items():
-            span.set_data(key, value)
     except Exception:
         pass
 
-    if span is None:
-        return nullcontext()
-    return span
+    return nullcontext()
 
 
 def start_span(
     name: str, op: str, origin: str, attributes: dict[str, Any]
-) -> Span | StreamedSpan | ContextManager[Any]:
-    """Start a span in the currently active trace lifecycle."""
+) -> StreamedSpan | ContextManager[Any]:
+    """Start a child span of the currently active span."""
     try:
-        is_span_streaming = has_span_streaming_enabled(sentry_sdk.get_client().options)
-
         # Early return to avoid large increase in span volume.
-        # Mirrors transaction-based tracing, in which `start_span()` no-ops when there is no active transaction.
-        if is_span_streaming and sentry_sdk.traces.get_current_span() is None:
+        # Only create spans when there is an active trace, e.g. a task spawned from a request.
+        if sentry_sdk.traces.get_current_span() is None:
             return nullcontext()
 
-        if is_span_streaming:
-            return sentry_sdk.traces.start_span(
-                name=name,
-                attributes={
-                    "sentry.op": op,
-                    **attributes,
-                },
-            )
-
-        span = sentry_sdk.start_span(
-            op=op,
+        return sentry_sdk.traces.start_span(
             name=name,
-            origin=origin,
+            attributes={
+                "sentry.op": op,
+                "sentry.origin": origin,
+                **attributes,
+            },
         )
-        for key, value in attributes.items():
-            span.set_data(key, value)
-
-        return span
     except Exception:
         pass
 

clients/python/src/taskbroker_client/worker/workerchild.py

@@ -32,12 +32,11 @@
 )
 from sentry_sdk.consts import OP, SPANDATA
 from sentry_sdk.crons import MonitorStatus, capture_checkin
-from sentry_sdk.tracing import Span
 
 from taskbroker_client.app import import_app
 from taskbroker_client.constants import CompressionType
 from taskbroker_client.retry import NoRetriesRemainingError
-from taskbroker_client.sdk import start_span, start_transaction
+from taskbroker_client.sdk import start_segment, start_span
 from taskbroker_client.state import clear_current_task, current_task, set_current_task
 from taskbroker_client.task import Task
 from taskbroker_client.types import ContextHook, InflightTaskActivation, ProcessingResult
@@ -669,7 +668,7 @@ def _execute_activation(
                 tags={"namespace": activation.namespace, "taskname": activation.taskname},
             ),
             sentry_sdk.isolation_scope(),
-            start_transaction(
+            start_segment(
                 name=activation.taskname,
                 op="queue.task.taskworker",
                 origin="taskworker",
@@ -682,13 +681,8 @@ def _execute_activation(
                         "task": activation.taskname,
                     }
                 },
-            ) as transaction,
+            ),
         ):
-            # Do not attach on StreamedSpan because eager serialization increases memory use.
-            if isinstance(transaction, Span):
-                transaction.set_data("taskworker-task.args", args)
-                transaction.set_data("taskworker-task.kwargs", kwargs)
-
             # ToDatetime().timestamp() reads a naive UTC datetime as local time.
             task_added_time = activation.received_at.seconds + activation.received_at.nanos / 1e9
             # latency attribute needs to be in milliseconds

clients/python/tests/test_task.py

@@ -302,7 +302,7 @@ def test_create_activation_tracing(task_namespace: TaskNamespace) -> None:
     def with_parameters(one: str, two: int, org_id: int) -> None:
         raise NotImplementedError
 
-    with sentry_sdk.start_transaction(op="test.task"):
+    with sentry_sdk.traces.start_span(name="test.task"):
         activation = with_parameters.create_activation(["one", 22], {"org_id": 99})
 
     headers = activation.headers
@@ -315,7 +315,7 @@ def test_create_activation_tracing_headers(task_namespace: TaskNamespace) -> Non
     def with_parameters(one: str, two: int, org_id: int) -> None:
         raise NotImplementedError
 
-    with sentry_sdk.start_transaction(op="test.task"):
+    with sentry_sdk.traces.start_span(name="test.task"):
         activation = with_parameters.create_activation(
             ["one", 22], {"org_id": 99}, {"key": "value"}
         )
@@ -331,7 +331,7 @@ def test_create_activation_tracing_disable(task_namespace: TaskNamespace) -> Non
     def with_parameters(one: str, two: int, org_id: int) -> None:
         raise NotImplementedError
 
-    with sentry_sdk.start_transaction(op="test.task"):
+    with sentry_sdk.traces.start_span(name="test.task"):
         activation = with_parameters.create_activation(
             ["one", 22], {"org_id": 99}, {"sentry-propagate-traces": False}
         )

clients/python/tests/worker/test_worker.py

@@ -13,7 +13,7 @@
 from multiprocessing import Event, get_context
 from multiprocessing.synchronize import Event as MultiprocessingEvent
 from pathlib import Path
-from typing import Any, Callable, Literal
+from typing import Any, Callable
 from unittest import TestCase, mock
 from uuid import uuid4
 
@@ -2679,7 +2679,6 @@ def test_child_process_terminate_task(mock_logger: mock.Mock) -> None:
 
 
 @pytest.mark.usefixtures("restore_signal_handlers")
-@pytest.mark.parametrize("trace_lifecycle", ["static", "stream"])
 @pytest.mark.parametrize(
     ("taskname", "args", "deadline", "retry_state"),
     [
@@ -2695,7 +2694,6 @@ def test_child_process_terminate_task(mock_logger: mock.Mock) -> None:
     ],
 )
 def test_child_process_error_event_uses_task_transaction(
-    trace_lifecycle: Literal["static", "stream"],
     taskname: str,
     args: list[Any],
     deadline: int,
@@ -2728,7 +2726,7 @@ def test_child_process_error_event_uses_task_transaction(
             transport=mock.Mock(spec=Transport),
             before_send=capture_event,
             traces_sample_rate=0,
-            trace_lifecycle=trace_lifecycle,
+            trace_lifecycle="stream",
         ) as client,
         sentry_sdk.isolation_scope() as scope,
     ):