From 87dcf25f8684e6c489c3845910f4b3974bd55e10 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 10:59:22 +0200 Subject: [PATCH 01/30] ref(o11y): Support SDK configured with the streaming trace lifecycle --- clients/python/pyproject.toml | 2 +- .../taskbroker_client/worker/workerchild.py | 91 ++++++++++++++----- uv.lock | 8 +- 3 files changed, 71 insertions(+), 30 deletions(-) diff --git a/clients/python/pyproject.toml b/clients/python/pyproject.toml index 92656511..f15b552d 100644 --- a/clients/python/pyproject.toml +++ b/clients/python/pyproject.toml @@ -5,7 +5,7 @@ description = "Taskbroker python client and worker runtime" readme = "README.md" dependencies = [ "sentry-arroyo>=2.41.0", - "sentry-sdk[http2]>=2.43.0", + "sentry-sdk[http2]>=2.52.0", "sentry-protos>=0.26.1", "confluent_kafka>=2.3.0", "cronsim>=2.6", diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py index d8f206e1..161b5f2f 100644 --- a/clients/python/src/taskbroker_client/worker/workerchild.py +++ b/clients/python/src/taskbroker_client/worker/workerchild.py @@ -14,6 +14,7 @@ from types import FrameType from typing import Any, Literal from uuid import UUID +from contextlib import contextmanager # XXX: Don't import any modules that will import django here, do those within child_process import msgpack @@ -32,6 +33,9 @@ ) from sentry_sdk.consts import OP, SPANDATA, SPANSTATUS from sentry_sdk.crons import MonitorStatus, capture_checkin +from sentry_sdk.tracing_utils import has_span_streaming_enabled +from sentry_sdk.scope import Scope +from sentry_sdk.tracing import Span from taskbroker_client.app import import_app from taskbroker_client.constants import CompressionType @@ -600,6 +604,25 @@ def check_task_future_completion( for task in pending_task_futures.copy(): await_task_futures(task) + @contextmanager + def _task_processing_span(activation: TaskActivation, latency: float) -> Span: + """Provide a span in the transaction-based tracing API with relevant attributes set.""" + with sentry_sdk.start_span( + op=OP.QUEUE_PROCESS, + name=activation.taskname, + origin="taskworker", + ) as span: + span.set_data(SPANDATA.MESSAGING_DESTINATION_NAME, activation.namespace) + span.set_data(SPANDATA.MESSAGING_MESSAGE_ID, activation.id) + span.set_data(SPANDATA.MESSAGING_MESSAGE_RECEIVE_LATENCY, latency) + span.set_data( + SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT, + activation.retry_state.attempts, + ) + span.set_data(SPANDATA.MESSAGING_SYSTEM, "taskworker") + yield span + + def _execute_activation( task_func: Task[Any, Any], activation: TaskActivation, @@ -611,46 +634,66 @@ def _execute_activation( args = parameters.get("args", []) kwargs = parameters.get("kwargs", {}) + is_span_streaming = has_span_streaming_enabled(sentry_sdk.get_client().options) + headers = dict(activation.headers) - transaction = sentry_sdk.continue_trace( - environ_or_headers=headers, - op="queue.task.taskworker", - name=activation.taskname, - origin="taskworker", - ) sampling_context = { "taskworker": { "task": activation.taskname, } } + if is_span_streaming: + transaction = sentry_sdk.continue_trace( + environ_or_headers=headers, + op="queue.task.taskworker", + name=activation.taskname, + origin="taskworker", + ) + + parent_span = sentry_sdk.start_transaction(transaction, custom_sampling_context=sampling_context) + else: + sentry_sdk.traces.continue_trace(headers) + Scope.set_custom_sampling_context(sampling_context) + + parent_span = sentry_sdk.traces.start_span( + name=activation.taskname, + attributes = { + "sentry.op": "queue.task.taskworker", + "sentry.origin": "taskworker", + } + ) + with ( metrics.track_memory_usage( "taskworker.worker.memory_change", tags={"namespace": activation.namespace, "taskname": activation.taskname}, ), sentry_sdk.isolation_scope(), - sentry_sdk.start_transaction(transaction, custom_sampling_context=sampling_context), + parent_span, ): - transaction.set_data( - "taskworker-task", {"args": args, "kwargs": kwargs, "id": activation.id} - ) + if not is_span_streaming: + parent_span.set_data( + "taskworker-task", {"args": args, "kwargs": kwargs, "id": activation.id} + ) + task_added_time = activation.received_at.ToDatetime().timestamp() # latency attribute needs to be in milliseconds latency = (time.time() - task_added_time) * 1000 - with sentry_sdk.start_span( - op=OP.QUEUE_PROCESS, - name=activation.taskname, - origin="taskworker", - ) as span: - span.set_data(SPANDATA.MESSAGING_DESTINATION_NAME, activation.namespace) - span.set_data(SPANDATA.MESSAGING_MESSAGE_ID, activation.id) - span.set_data(SPANDATA.MESSAGING_MESSAGE_RECEIVE_LATENCY, latency) - span.set_data( - SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT, activation.retry_state.attempts - ) - span.set_data(SPANDATA.MESSAGING_SYSTEM, "taskworker") - + child_span = sentry_sdk.traces.start_span( + activation.taskname, + attributes={ + "sentry.op": OP.QUEUE_PROCESS, + "sentry.origin": "taskworker", + SPANDATA.MESSAGING_DESTINATION_NAME: activation.namespace, + SPANDATA.MESSAGING_MESSAGE_ID: activation.id, + SPANDATA.MESSAGING_MESSAGE_RECEIVE_LATENCY: latency, + SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT: activation.retry_state.attempts, + SPANDATA.MESSAGING_SYSTEM: "taskworker", + } + ) if is_span_streaming else _task_processing_span(activation=activation, latency=latency) + + with child_span: # TODO(taskworker) remove this when doing cleanup # The `__start_time` parameter is spliced into task parameters by # sentry.celery.SentryTask._add_metadata and needs to be removed @@ -678,9 +721,7 @@ def _execute_activation( task_func(*args, headers=headers, **kwargs) else: task_func(*args, **kwargs) - transaction.set_status(SPANSTATUS.OK) except Exception: - transaction.set_status(SPANSTATUS.INTERNAL_ERROR) raise def record_task_execution( diff --git a/uv.lock b/uv.lock index 4776d70b..b0e85a2f 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 2 +revision = 3 requires-python = ">=3.11" resolution-markers = [ "sys_platform == 'darwin' or sys_platform == 'linux'", @@ -703,14 +703,14 @@ wheels = [ [[package]] name = "sentry-sdk" -version = "2.49.0" +version = "2.66.0" source = { registry = "https://pypi.devinfra.sentry.io/simple" } dependencies = [ { name = "certifi", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" }, { name = "urllib3", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" }, ] wheels = [ - { url = "https://pypi.devinfra.sentry.io/wheels/sentry_sdk-2.49.0-py2.py3-none-any.whl", hash = "sha256:6ea78499133874445a20fe9c826c9e960070abeb7ae0cdf930314ab16bb97aa0" }, + { url = "https://pypi.devinfra.sentry.io/wheels/sentry_sdk-2.66.0-py3-none-any.whl", hash = "sha256:096136c214c602be2b323524d30755dc5b30ec5a218a206207f33b12c05c6f11" }, ] [package.optional-dependencies] @@ -856,7 +856,7 @@ requires-dist = [ { name = "redis-py-cluster", marker = "extra == 'cluster'", specifier = ">=2.1.0" }, { name = "sentry-arroyo", specifier = ">=2.41.0" }, { name = "sentry-protos", specifier = ">=0.26.1" }, - { name = "sentry-sdk", extras = ["http2"], specifier = ">=2.43.0" }, + { name = "sentry-sdk", extras = ["http2"], specifier = ">=2.52.0" }, { name = "setuptools", marker = "extra == 'examples'", specifier = ">=80.0" }, { name = "zstandard", specifier = ">=0.18.0" }, ] From 99ff2dea2a3d10ccb0d685326d73ae7751869842 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 11:15:05 +0200 Subject: [PATCH 02/30] python lint --- .../taskbroker_client/worker/workerchild.py | 53 ++++++++++--------- 1 file changed, 29 insertions(+), 24 deletions(-) diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py index 161b5f2f..1937cfd0 100644 --- a/clients/python/src/taskbroker_client/worker/workerchild.py +++ b/clients/python/src/taskbroker_client/worker/workerchild.py @@ -622,7 +622,6 @@ def _task_processing_span(activation: TaskActivation, latency: float) -> Span: span.set_data(SPANDATA.MESSAGING_SYSTEM, "taskworker") yield span - def _execute_activation( task_func: Task[Any, Any], activation: TaskActivation, @@ -643,6 +642,17 @@ def _execute_activation( } } if is_span_streaming: + sentry_sdk.traces.continue_trace(headers) + Scope.set_custom_sampling_context(sampling_context) + + parent_span = sentry_sdk.traces.start_span( + name=activation.taskname, + attributes={ + "sentry.op": "queue.task.taskworker", + "sentry.origin": "taskworker", + }, + ) + else: transaction = sentry_sdk.continue_trace( environ_or_headers=headers, op="queue.task.taskworker", @@ -650,17 +660,8 @@ def _execute_activation( origin="taskworker", ) - parent_span = sentry_sdk.start_transaction(transaction, custom_sampling_context=sampling_context) - else: - sentry_sdk.traces.continue_trace(headers) - Scope.set_custom_sampling_context(sampling_context) - - parent_span = sentry_sdk.traces.start_span( - name=activation.taskname, - attributes = { - "sentry.op": "queue.task.taskworker", - "sentry.origin": "taskworker", - } + parent_span = sentry_sdk.start_transaction( + transaction, custom_sampling_context=sampling_context ) with ( @@ -680,18 +681,22 @@ def _execute_activation( # latency attribute needs to be in milliseconds latency = (time.time() - task_added_time) * 1000 - child_span = sentry_sdk.traces.start_span( - activation.taskname, - attributes={ - "sentry.op": OP.QUEUE_PROCESS, - "sentry.origin": "taskworker", - SPANDATA.MESSAGING_DESTINATION_NAME: activation.namespace, - SPANDATA.MESSAGING_MESSAGE_ID: activation.id, - SPANDATA.MESSAGING_MESSAGE_RECEIVE_LATENCY: latency, - SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT: activation.retry_state.attempts, - SPANDATA.MESSAGING_SYSTEM: "taskworker", - } - ) if is_span_streaming else _task_processing_span(activation=activation, latency=latency) + child_span = ( + sentry_sdk.traces.start_span( + activation.taskname, + attributes={ + "sentry.op": OP.QUEUE_PROCESS, + "sentry.origin": "taskworker", + SPANDATA.MESSAGING_DESTINATION_NAME: activation.namespace, + SPANDATA.MESSAGING_MESSAGE_ID: activation.id, + SPANDATA.MESSAGING_MESSAGE_RECEIVE_LATENCY: latency, + SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT: activation.retry_state.attempts, + SPANDATA.MESSAGING_SYSTEM: "taskworker", + }, + ) + if is_span_streaming + else _task_processing_span(activation=activation, latency=latency) + ) with child_span: # TODO(taskworker) remove this when doing cleanup From 6c06b23fa1140ceff27302585b3e41c6d0e3bd03 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 11:21:32 +0200 Subject: [PATCH 03/30] make mypy happy --- .../python/src/taskbroker_client/worker/workerchild.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py index 1937cfd0..5ea7df70 100644 --- a/clients/python/src/taskbroker_client/worker/workerchild.py +++ b/clients/python/src/taskbroker_client/worker/workerchild.py @@ -14,7 +14,7 @@ from types import FrameType from typing import Any, Literal from uuid import UUID -from contextlib import contextmanager +from contextlib import contextmanager, AbstractContextManager # XXX: Don't import any modules that will import django here, do those within child_process import msgpack @@ -605,7 +605,7 @@ def check_task_future_completion( await_task_futures(task) @contextmanager - def _task_processing_span(activation: TaskActivation, latency: float) -> Span: + def _task_processing_span(activation: TaskActivation, latency: float) -> Generator[Span, None, None]: """Provide a span in the transaction-based tracing API with relevant attributes set.""" with sentry_sdk.start_span( op=OP.QUEUE_PROCESS, @@ -672,7 +672,7 @@ def _execute_activation( sentry_sdk.isolation_scope(), parent_span, ): - if not is_span_streaming: + if isinstance(parent_span, Span): parent_span.set_data( "taskworker-task", {"args": args, "kwargs": kwargs, "id": activation.id} ) @@ -681,7 +681,7 @@ def _execute_activation( # latency attribute needs to be in milliseconds latency = (time.time() - task_added_time) * 1000 - child_span = ( + child_span: AbstractContextManager[Any] = ( sentry_sdk.traces.start_span( activation.taskname, attributes={ From 43bd7ee19be7dc334bcd2545883447aebb617a45 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 11:26:17 +0200 Subject: [PATCH 04/30] add missing declaration --- .../src/taskbroker_client/worker/workerchild.py | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py index 5ea7df70..50ecfd1e 100644 --- a/clients/python/src/taskbroker_client/worker/workerchild.py +++ b/clients/python/src/taskbroker_client/worker/workerchild.py @@ -8,13 +8,13 @@ import threading import time from collections.abc import Callable, Generator, Sequence +from contextlib import AbstractContextManager, contextmanager from dataclasses import dataclass from functools import partial from multiprocessing.synchronize import Event from types import FrameType from typing import Any, Literal from uuid import UUID -from contextlib import contextmanager, AbstractContextManager # XXX: Don't import any modules that will import django here, do those within child_process import msgpack @@ -31,11 +31,12 @@ TaskActivation, TaskActivationStatus, ) -from sentry_sdk.consts import OP, SPANDATA, SPANSTATUS +from sentry_sdk.consts import OP, SPANDATA from sentry_sdk.crons import MonitorStatus, capture_checkin -from sentry_sdk.tracing_utils import has_span_streaming_enabled from sentry_sdk.scope import Scope -from sentry_sdk.tracing import Span +from sentry_sdk.traces import StreamedSpan +from sentry_sdk.tracing import NoOpSpan, Span, Transaction +from sentry_sdk.tracing_utils import has_span_streaming_enabled from taskbroker_client.app import import_app from taskbroker_client.constants import CompressionType @@ -605,7 +606,9 @@ def check_task_future_completion( await_task_futures(task) @contextmanager - def _task_processing_span(activation: TaskActivation, latency: float) -> Generator[Span, None, None]: + def _task_processing_span( + activation: TaskActivation, latency: float + ) -> Generator[Span, None, None]: """Provide a span in the transaction-based tracing API with relevant attributes set.""" with sentry_sdk.start_span( op=OP.QUEUE_PROCESS, @@ -641,6 +644,8 @@ def _execute_activation( "task": activation.taskname, } } + + parent_span: Transaction | NoOpSpan | StreamedSpan if is_span_streaming: sentry_sdk.traces.continue_trace(headers) Scope.set_custom_sampling_context(sampling_context) From 1e907343c8d63b8387787f43ec6069d45a349770 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 13:12:48 +0200 Subject: [PATCH 05/30] test: Initialize SDK in relevant tests --- clients/python/tests/conftest.py | 37 ++++++ clients/python/tests/test_task.py | 20 ++- clients/python/tests/worker/test_worker.py | 135 +++++++++++++++++---- 3 files changed, 165 insertions(+), 27 deletions(-) diff --git a/clients/python/tests/conftest.py b/clients/python/tests/conftest.py index 2d777a25..a43f7ae0 100644 --- a/clients/python/tests/conftest.py +++ b/clients/python/tests/conftest.py @@ -1,7 +1,14 @@ +from collections.abc import Callable, Generator from datetime import UTC, datetime +from typing import Any +import pytest +import sentry_sdk import time_machine from arroyo.backends.kafka import KafkaProducer +from pytest import FixtureRequest +from sentry_sdk.envelope import Envelope +from sentry_sdk.transport import Transport from taskbroker_client.types import AtMostOnceStore @@ -21,6 +28,36 @@ def freeze_time(t: str | datetime | None = None) -> time_machine.travel: return time_machine.travel(t, tick=False) +@pytest.fixture +def sentry_init(request: FixtureRequest) -> Generator[Callable[..., None], None, None]: + def inner(*a: Any, **kw: Any) -> None: + kw.setdefault("transport", TestTransport()) + client = sentry_sdk.Client(*a, **kw) + sentry_sdk.get_global_scope().set_client(client) + + if request.node.get_closest_marker("forked"): + # Do not run isolation if the test is already running in + # ultimate isolation (seems to be required for celery tests that + # fork) + yield inner + else: + old_client = sentry_sdk.get_global_scope().client + try: + sentry_sdk.get_current_scope().set_client(None) + yield inner + finally: + sentry_sdk.get_global_scope().set_client(old_client) + + +class TestTransport(Transport): + def __init__(self) -> None: + Transport.__init__(self) + + def capture_envelope(self, _: Envelope) -> None: + """No-op capture_envelope for tests""" + pass + + class StubAtMostOnce(AtMostOnceStore): def __init__(self) -> None: self._keys: dict[str, str] = {} diff --git a/clients/python/tests/test_task.py b/clients/python/tests/test_task.py index 2556b9e7..c9e8501e 100644 --- a/clients/python/tests/test_task.py +++ b/clients/python/tests/test_task.py @@ -2,7 +2,7 @@ import datetime from collections.abc import MutableMapping from concurrent.futures import Future -from typing import Any +from typing import Any, Callable from unittest.mock import patch import msgpack @@ -297,7 +297,11 @@ def with_parameters(one: str, two: int, org_id: int) -> None: assert activation.parameters == "" -def test_create_activation_tracing(task_namespace: TaskNamespace) -> None: +def test_create_activation_tracing( + sentry_init: Callable[..., None], task_namespace: TaskNamespace +) -> None: + sentry_init(traces_sample_rate=1.0) + @task_namespace.register(name="test.parameters") def with_parameters(one: str, two: int, org_id: int) -> None: raise NotImplementedError @@ -310,7 +314,11 @@ def with_parameters(one: str, two: int, org_id: int) -> None: assert "baggage" in headers -def test_create_activation_tracing_headers(task_namespace: TaskNamespace) -> None: +def test_create_activation_tracing_headers( + sentry_init: Callable[..., None], task_namespace: TaskNamespace +) -> None: + sentry_init(traces_sample_rate=1.0) + @task_namespace.register(name="test.parameters") def with_parameters(one: str, two: int, org_id: int) -> None: raise NotImplementedError @@ -326,7 +334,11 @@ def with_parameters(one: str, two: int, org_id: int) -> None: assert headers["key"] == "value" -def test_create_activation_tracing_disable(task_namespace: TaskNamespace) -> None: +def test_create_activation_tracing_disable( + sentry_init: Callable[..., None], task_namespace: TaskNamespace +) -> None: + sentry_init(traces_sample_rate=1.0) + @task_namespace.register(name="test.parameters") def with_parameters(one: str, two: int, org_id: int) -> None: raise NotImplementedError diff --git a/clients/python/tests/worker/test_worker.py b/clients/python/tests/worker/test_worker.py index 7253a74e..5769228d 100644 --- a/clients/python/tests/worker/test_worker.py +++ b/clients/python/tests/worker/test_worker.py @@ -1022,7 +1022,11 @@ def test_push_task_worker_busy(self) -> None: @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") -def test_child_process_complete(mock_capture_checkin: mock.MagicMock) -> None: +def test_child_process_complete( + sentry_init: Callable[..., None], mock_capture_checkin: mock.MagicMock +) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1047,7 +1051,11 @@ def test_child_process_complete(mock_capture_checkin: mock.MagicMock) -> None: assert mock_capture_checkin.call_count == 0 -def test_child_process_canary_task(capsys: pytest.CaptureFixture[str]) -> None: +def test_child_process_canary_task( + sentry_init: Callable[..., None], capsys: pytest.CaptureFixture[str] +) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1073,7 +1081,9 @@ def test_child_process_canary_task(capsys: pytest.CaptureFixture[str]) -> None: assert capsys.readouterr().out == "Done running canary task!\n" -def test_child_process_emits_running_message() -> None: +def test_child_process_emits_running_message(sentry_init: Callable[..., None]) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1106,8 +1116,11 @@ def test_child_process_emits_running_message() -> None: @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") def test_child_process_emits_exiting_once_and_continues_until_release( + sentry_init: Callable[..., None], mock_capture_checkin: mock.MagicMock, ) -> None: + sentry_init(traces_sample_rate=1.0) + shutdown = Event() ctx = get_context("fork") child_id = uuid4() @@ -1167,7 +1180,9 @@ def test_child_process_emits_exiting_once_and_continues_until_release( assert mock_capture_checkin.call_count == 0 -def test_child_process_emits_busy_and_idle_messages() -> None: +def test_child_process_emits_busy_and_idle_messages(sentry_init: Callable[..., None]) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1199,7 +1214,9 @@ def test_child_process_emits_busy_and_idle_messages() -> None: assert processed.get(timeout=1).task_id == SIMPLE_TASK.activation.id -def test_child_process_remove_start_time_kwargs() -> None: +def test_child_process_remove_start_time_kwargs(sentry_init: Callable[..., None]) -> None: + sentry_init(traces_sample_rate=1.0) + activation = InflightTaskActivation( host="localhost:50051", receive_timestamp=0, @@ -1236,7 +1253,9 @@ def test_child_process_remove_start_time_kwargs() -> None: assert result.status == TASK_ACTIVATION_STATUS_COMPLETE -def test_child_process_retry_task() -> None: +def test_child_process_retry_task(sentry_init: Callable[..., None]) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1263,8 +1282,10 @@ def test_child_process_retry_task() -> None: @mock.patch("taskbroker_client.worker.workerchild.logger") @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") def test_child_process_retry_task_max_attempts( - mock_capture: mock.Mock, mock_logger: mock.Mock + sentry_init: Callable[..., None], mock_capture: mock.Mock, mock_logger: mock.Mock ) -> None: + sentry_init(traces_sample_rate=1.0) + # Create an activation that is on its final attempt and # will raise an error again. activation = InflightTaskActivation( @@ -1324,7 +1345,9 @@ def test_child_process_retry_task_max_attempts( assert extra["retry_max_attempts"] == 3 -def test_child_process_failure_task() -> None: +def test_child_process_failure_task(sentry_init: Callable[..., None]) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1348,7 +1371,9 @@ def test_child_process_failure_task() -> None: assert result.status == TASK_ACTIVATION_STATUS_FAILURE -def test_child_process_shutdown() -> None: +def test_child_process_shutdown(sentry_init: Callable[..., None]) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1372,7 +1397,9 @@ def test_child_process_shutdown() -> None: assert processed.qsize() == 0 -def test_child_process_unknown_task() -> None: +def test_child_process_unknown_task(sentry_init: Callable[..., None]) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1400,7 +1427,9 @@ def test_child_process_unknown_task() -> None: assert result.status == TASK_ACTIVATION_STATUS_COMPLETE -def test_child_process_at_most_once() -> None: +def test_child_process_at_most_once(sentry_init: Callable[..., None]) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1431,7 +1460,11 @@ def test_child_process_at_most_once() -> None: @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") -def test_child_process_record_checkin(mock_capture_checkin: mock.Mock) -> None: +def test_child_process_record_checkin( + sentry_init: Callable[..., None], mock_capture_checkin: mock.Mock +) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1463,8 +1496,10 @@ def test_child_process_record_checkin(mock_capture_checkin: mock.Mock) -> None: ) -def test_child_process_pass_headers() -> None: +def test_child_process_pass_headers(sentry_init: Callable[..., None]) -> None: """Task with pass_headers=True receives headers from the activation.""" + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1494,7 +1529,11 @@ def test_child_process_pass_headers() -> None: @mock.patch("taskbroker_client.worker.workerchild.logger") -def test_child_process_terminate_task(mock_logger: mock.Mock) -> None: +def test_child_process_terminate_task( + sentry_init: Callable[..., None], mock_logger: mock.Mock +) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1541,7 +1580,10 @@ def test_child_process_terminate_task(mock_logger: mock.Mock) -> None: @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") -def test_child_process_decompression(mock_capture_checkin: mock.MagicMock) -> None: +def test_child_process_decompression( + sentry_init: Callable[..., None], mock_capture_checkin: mock.MagicMock +) -> None: + sentry_init(traces_sample_rate=1.0) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1567,8 +1609,10 @@ def test_child_process_decompression(mock_capture_checkin: mock.MagicMock) -> No assert mock_capture_checkin.call_count == 0 -def test_child_process_context_hooks() -> None: +def test_child_process_context_hooks(sentry_init: Callable[..., None]) -> None: """Context hooks' on_execute is called with activation headers during task execution.""" + sentry_init(traces_sample_rate=1.0) + executed_headers: list[dict[str, str]] = [] class RecordingHook: @@ -1625,7 +1669,11 @@ def on_execute(self, headers: dict[str, str]) -> contextlib.AbstractContextManag @mock.patch("taskbroker_client.worker.workerchild.logger") -def test_child_process_silenced_timeout(mock_logger: mock.Mock) -> None: +def test_child_process_silenced_timeout( + sentry_init: Callable[..., None], mock_logger: mock.Mock +) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1656,7 +1704,11 @@ def test_child_process_silenced_timeout(mock_logger: mock.Mock) -> None: @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") -def test_child_process_silenced_exception_with_retries(mock_capture: mock.Mock) -> None: +def test_child_process_silenced_exception_with_retries( + sentry_init: Callable[..., None], mock_capture: mock.Mock +) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1684,7 +1736,11 @@ def test_child_process_silenced_exception_with_retries(mock_capture: mock.Mock) @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") -def test_child_process_expected_ignored_exception_max_attempts(mock_capture: mock.Mock) -> None: +def test_child_process_expected_ignored_exception_max_attempts( + sentry_init: Callable[..., None], mock_capture: mock.Mock +) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1714,9 +1770,11 @@ def test_child_process_expected_ignored_exception_max_attempts(mock_capture: moc @mock.patch("taskbroker_client.worker.workerchild.logger") @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") def test_child_process_silenced_exception_max_attempts( - mock_capture: mock.Mock, mock_logger: mock.Mock + sentry_init: Callable[..., None], mock_capture: mock.Mock, mock_logger: mock.Mock ) -> None: """Silenced exceptions do not raise on retry exhaustion.""" + sentry_init(traces_sample_rate=1.0) + activation = InflightTaskActivation( host="localhost:50051", receive_timestamp=0, @@ -1768,7 +1826,11 @@ def test_child_process_silenced_exception_max_attempts( @mock.patch("taskbroker_client.worker.workerchild.logger") -def test_child_process_retry_on_deadline_exceeded(mock_logger: mock.Mock) -> None: +def test_child_process_retry_on_deadline_exceeded( + sentry_init: Callable[..., None], mock_logger: mock.Mock +) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1801,8 +1863,12 @@ def test_child_process_retry_on_deadline_exceeded(mock_logger: mock.Mock) -> Non @mock.patch("taskbroker_client.worker.workerchild.logger") -def test_child_process_general_exception_logs_task_failed(mock_logger: mock.Mock) -> None: +def test_child_process_general_exception_logs_task_failed( + sentry_init: Callable[..., None], mock_logger: mock.Mock +) -> None: """A non-retriable Exception emits taskworker.task.failed with all fields.""" + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1838,10 +1904,13 @@ def test_child_process_general_exception_logs_task_failed(mock_logger: mock.Mock @mock.patch("taskbroker_client.worker.workerchild.logger") def test_child_process_silenced_exception_does_not_log_task_failed( + sentry_init: Callable[..., None], mock_logger: mock.Mock, ) -> None: """When err is in silenced_exceptions, taskworker.task.failed is NOT logged. Preserves the silencing semantics added in #608.""" + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1936,10 +2005,13 @@ def _producing_task(task_id: str = "task-with-futures") -> InflightTaskActivatio @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_tracks_producer_futures( + sentry_init: Callable[..., None], producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: + sentry_init(traces_sample_rate=1.0) + task = _producing_task() todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1974,10 +2046,13 @@ def test_child_process_tracks_producer_futures( @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_holds_result_until_futures_done( + sentry_init: Callable[..., None], producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: + sentry_init(traces_sample_rate=1.0) + task = _producing_task() todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -2029,10 +2104,13 @@ def observe_and_resolve() -> None: @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_skip_awaiting_futures_places_result_immediately( + sentry_init: Callable[..., None], producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: + sentry_init(traces_sample_rate=1.0) + task = _producing_task() todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -2091,10 +2169,13 @@ def observe_and_resolve() -> None: @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_drains_pending_futures_on_sigterm( + sentry_init: Callable[..., None], producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: + sentry_init(traces_sample_rate=1.0) + task = _producing_task() todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -2139,10 +2220,13 @@ def deliver_sigterm() -> None: @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_retries_on_failed_future( + sentry_init: Callable[..., None], producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: + sentry_init(traces_sample_rate=1.0) + retriable_task = InflightTaskActivation( host="localhost:50051", receive_timestamp=0, @@ -2189,10 +2273,13 @@ def test_child_process_retries_on_failed_future( @pytest.mark.parametrize("pending_registry", _PENDING_REGISTRIES) def test_child_process_clears_pending_futures_when_task_fails( + sentry_init: Callable[..., None], pending_registry: Any, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: + sentry_init(traces_sample_rate=1.0) + leftover_future: Future[BrokerValue[KafkaPayload]] = Future() leftover_future.set_result(_make_broker_value()) pending_registry["test.producer"].append(leftover_future) @@ -2226,9 +2313,11 @@ def test_child_process_clears_pending_futures_when_task_fails( def test_child_process_uses_configured_future_checking_frequency( - clear_pending_futures: None, restore_signal_handlers: None + sentry_init: Callable[..., None], clear_pending_futures: None, restore_signal_handlers: None ) -> None: """The idle future-checking loop polls on the configured interval.""" + sentry_init(traces_sample_rate=1.0) + # A task that runs long enough for the idle future-checking loop to poll a # few times before max_task_count triggers shutdown. slow_task = InflightTaskActivation( From 5de3e0dd777d5b8b5a843277ac8951ffc57ed95a Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 13:13:33 +0200 Subject: [PATCH 06/30] remove dead code --- clients/python/tests/conftest.py | 16 +++++----------- 1 file changed, 5 insertions(+), 11 deletions(-) diff --git a/clients/python/tests/conftest.py b/clients/python/tests/conftest.py index a43f7ae0..a4ad61a8 100644 --- a/clients/python/tests/conftest.py +++ b/clients/python/tests/conftest.py @@ -35,18 +35,12 @@ def inner(*a: Any, **kw: Any) -> None: client = sentry_sdk.Client(*a, **kw) sentry_sdk.get_global_scope().set_client(client) - if request.node.get_closest_marker("forked"): - # Do not run isolation if the test is already running in - # ultimate isolation (seems to be required for celery tests that - # fork) + old_client = sentry_sdk.get_global_scope().client + try: + sentry_sdk.get_current_scope().set_client(None) yield inner - else: - old_client = sentry_sdk.get_global_scope().client - try: - sentry_sdk.get_current_scope().set_client(None) - yield inner - finally: - sentry_sdk.get_global_scope().set_client(old_client) + finally: + sentry_sdk.get_global_scope().set_client(old_client) class TestTransport(Transport): From d570be1cdb1411dde2c19064e4cf75f8decc54fa Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 13:26:26 +0200 Subject: [PATCH 07/30] ref(o11y): Remove redundant span status assignment --- .../taskbroker_client/worker/workerchild.py | 45 +++++++++---------- 1 file changed, 20 insertions(+), 25 deletions(-) diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py index d8f206e1..726b4722 100644 --- a/clients/python/src/taskbroker_client/worker/workerchild.py +++ b/clients/python/src/taskbroker_client/worker/workerchild.py @@ -30,7 +30,7 @@ TaskActivation, TaskActivationStatus, ) -from sentry_sdk.consts import OP, SPANDATA, SPANSTATUS +from sentry_sdk.consts import OP, SPANDATA from sentry_sdk.crons import MonitorStatus, capture_checkin from taskbroker_client.app import import_app @@ -658,30 +658,25 @@ def _execute_activation( if "__start_time" in kwargs: kwargs.pop("__start_time") - try: - with contextlib.ExitStack() as stack: - with metrics.timer( - "taskworker.worker.context_rebuild.duration", - tags={ - "namespace": activation.namespace, - "taskname": activation.taskname, - }, - ): - for hook in context_hooks: - stack.enter_context(hook.on_execute(headers)) - if task_func.pass_headers: - if "headers" in kwargs: - raise TypeError( - f"Task '{task_func.name}' has pass_headers=True, but 'headers' was passed in kwargs. " - "The 'headers' parameter is injected by the worker and cannot be passed by the caller." - ) - task_func(*args, headers=headers, **kwargs) - else: - task_func(*args, **kwargs) - transaction.set_status(SPANSTATUS.OK) - except Exception: - transaction.set_status(SPANSTATUS.INTERNAL_ERROR) - raise + with contextlib.ExitStack() as stack: + with metrics.timer( + "taskworker.worker.context_rebuild.duration", + tags={ + "namespace": activation.namespace, + "taskname": activation.taskname, + }, + ): + for hook in context_hooks: + stack.enter_context(hook.on_execute(headers)) + if task_func.pass_headers: + if "headers" in kwargs: + raise TypeError( + f"Task '{task_func.name}' has pass_headers=True, but 'headers' was passed in kwargs. " + "The 'headers' parameter is injected by the worker and cannot be passed by the caller." + ) + task_func(*args, headers=headers, **kwargs) + else: + task_func(*args, **kwargs) def record_task_execution( activation: TaskActivation, From 283096bc69719368dd5dcb022b74e9be3fd670d8 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 14:00:30 +0200 Subject: [PATCH 08/30] add pytest parametrization --- clients/python/tests/test_task.py | 24 +- clients/python/tests/worker/test_worker.py | 277 ++++++++++++++++----- 2 files changed, 239 insertions(+), 62 deletions(-) diff --git a/clients/python/tests/test_task.py b/clients/python/tests/test_task.py index c9e8501e..ec5baa11 100644 --- a/clients/python/tests/test_task.py +++ b/clients/python/tests/test_task.py @@ -297,10 +297,14 @@ def with_parameters(one: str, two: int, org_id: int) -> None: assert activation.parameters == "" +@pytest.mark.parametrize("span_streaming", (False, True)) def test_create_activation_tracing( - sentry_init: Callable[..., None], task_namespace: TaskNamespace + sentry_init: Callable[..., None], span_streaming: bool, task_namespace: TaskNamespace ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) @task_namespace.register(name="test.parameters") def with_parameters(one: str, two: int, org_id: int) -> None: @@ -314,10 +318,14 @@ def with_parameters(one: str, two: int, org_id: int) -> None: assert "baggage" in headers +@pytest.mark.parametrize("span_streaming", (False, True)) def test_create_activation_tracing_headers( - sentry_init: Callable[..., None], task_namespace: TaskNamespace + sentry_init: Callable[..., None], span_streaming: bool, task_namespace: TaskNamespace ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) @task_namespace.register(name="test.parameters") def with_parameters(one: str, two: int, org_id: int) -> None: @@ -334,10 +342,14 @@ def with_parameters(one: str, two: int, org_id: int) -> None: assert headers["key"] == "value" +@pytest.mark.parametrize("span_streaming", (False, True)) def test_create_activation_tracing_disable( - sentry_init: Callable[..., None], task_namespace: TaskNamespace + sentry_init: Callable[..., None], span_streaming: bool, task_namespace: TaskNamespace ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) @task_namespace.register(name="test.parameters") def with_parameters(one: str, two: int, org_id: int) -> None: diff --git a/clients/python/tests/worker/test_worker.py b/clients/python/tests/worker/test_worker.py index 5769228d..bfb15b27 100644 --- a/clients/python/tests/worker/test_worker.py +++ b/clients/python/tests/worker/test_worker.py @@ -1021,11 +1021,15 @@ def test_push_task_worker_busy(self) -> None: ) +@pytest.mark.parametrize("span_streaming", (False, True)) @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") def test_child_process_complete( - sentry_init: Callable[..., None], mock_capture_checkin: mock.MagicMock + sentry_init: Callable[..., None], span_streaming: bool, mock_capture_checkin: mock.MagicMock ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1051,10 +1055,14 @@ def test_child_process_complete( assert mock_capture_checkin.call_count == 0 +@pytest.mark.parametrize("span_streaming", (False, True)) def test_child_process_canary_task( - sentry_init: Callable[..., None], capsys: pytest.CaptureFixture[str] + sentry_init: Callable[..., None], span_streaming: bool, capsys: pytest.CaptureFixture[str] ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1081,8 +1089,14 @@ def test_child_process_canary_task( assert capsys.readouterr().out == "Done running canary task!\n" -def test_child_process_emits_running_message(sentry_init: Callable[..., None]) -> None: - sentry_init(traces_sample_rate=1.0) +@pytest.mark.parametrize("span_streaming", (False, True)) +def test_child_process_emits_running_message( + sentry_init: Callable[..., None], span_streaming: bool +) -> None: + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1114,12 +1128,17 @@ def test_child_process_emits_running_message(sentry_init: Callable[..., None]) - assert message == ChildMessage(child_id, "running") +@pytest.mark.parametrize("span_streaming", (False, True)) @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") def test_child_process_emits_exiting_once_and_continues_until_release( - sentry_init: Callable[..., None], mock_capture_checkin: mock.MagicMock, + sentry_init: Callable[..., None], + span_streaming: bool, ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) shutdown = Event() ctx = get_context("fork") @@ -1180,8 +1199,14 @@ def test_child_process_emits_exiting_once_and_continues_until_release( assert mock_capture_checkin.call_count == 0 -def test_child_process_emits_busy_and_idle_messages(sentry_init: Callable[..., None]) -> None: - sentry_init(traces_sample_rate=1.0) +@pytest.mark.parametrize("span_streaming", (False, True)) +def test_child_process_emits_busy_and_idle_messages( + sentry_init: Callable[..., None], span_streaming: bool +) -> None: + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1214,8 +1239,14 @@ def test_child_process_emits_busy_and_idle_messages(sentry_init: Callable[..., N assert processed.get(timeout=1).task_id == SIMPLE_TASK.activation.id -def test_child_process_remove_start_time_kwargs(sentry_init: Callable[..., None]) -> None: - sentry_init(traces_sample_rate=1.0) +@pytest.mark.parametrize("span_streaming", (False, True)) +def test_child_process_remove_start_time_kwargs( + sentry_init: Callable[..., None], span_streaming: bool +) -> None: + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) activation = InflightTaskActivation( host="localhost:50051", @@ -1253,8 +1284,12 @@ def test_child_process_remove_start_time_kwargs(sentry_init: Callable[..., None] assert result.status == TASK_ACTIVATION_STATUS_COMPLETE -def test_child_process_retry_task(sentry_init: Callable[..., None]) -> None: - sentry_init(traces_sample_rate=1.0) +@pytest.mark.parametrize("span_streaming", (False, True)) +def test_child_process_retry_task(sentry_init: Callable[..., None], span_streaming: bool) -> None: + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1279,12 +1314,19 @@ def test_child_process_retry_task(sentry_init: Callable[..., None]) -> None: assert result.status == TASK_ACTIVATION_STATUS_RETRY +@pytest.mark.parametrize("span_streaming", (False, True)) @mock.patch("taskbroker_client.worker.workerchild.logger") @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") def test_child_process_retry_task_max_attempts( - sentry_init: Callable[..., None], mock_capture: mock.Mock, mock_logger: mock.Mock + mock_capture: mock.Mock, + mock_logger: mock.Mock, + sentry_init: Callable[..., None], + span_streaming: bool, ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) # Create an activation that is on its final attempt and # will raise an error again. @@ -1345,8 +1387,12 @@ def test_child_process_retry_task_max_attempts( assert extra["retry_max_attempts"] == 3 -def test_child_process_failure_task(sentry_init: Callable[..., None]) -> None: - sentry_init(traces_sample_rate=1.0) +@pytest.mark.parametrize("span_streaming", (False, True)) +def test_child_process_failure_task(sentry_init: Callable[..., None], span_streaming: bool) -> None: + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1371,8 +1417,12 @@ def test_child_process_failure_task(sentry_init: Callable[..., None]) -> None: assert result.status == TASK_ACTIVATION_STATUS_FAILURE -def test_child_process_shutdown(sentry_init: Callable[..., None]) -> None: - sentry_init(traces_sample_rate=1.0) +@pytest.mark.parametrize("span_streaming", (False, True)) +def test_child_process_shutdown(sentry_init: Callable[..., None], span_streaming: bool) -> None: + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1397,8 +1447,12 @@ def test_child_process_shutdown(sentry_init: Callable[..., None]) -> None: assert processed.qsize() == 0 -def test_child_process_unknown_task(sentry_init: Callable[..., None]) -> None: - sentry_init(traces_sample_rate=1.0) +@pytest.mark.parametrize("span_streaming", (False, True)) +def test_child_process_unknown_task(sentry_init: Callable[..., None], span_streaming: bool) -> None: + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1427,8 +1481,12 @@ def test_child_process_unknown_task(sentry_init: Callable[..., None]) -> None: assert result.status == TASK_ACTIVATION_STATUS_COMPLETE -def test_child_process_at_most_once(sentry_init: Callable[..., None]) -> None: - sentry_init(traces_sample_rate=1.0) +@pytest.mark.parametrize("span_streaming", (False, True)) +def test_child_process_at_most_once(sentry_init: Callable[..., None], span_streaming: bool) -> None: + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1459,11 +1517,17 @@ def test_child_process_at_most_once(sentry_init: Callable[..., None]) -> None: assert result.status == TASK_ACTIVATION_STATUS_COMPLETE +@pytest.mark.parametrize("span_streaming", (False, True)) @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") def test_child_process_record_checkin( - sentry_init: Callable[..., None], mock_capture_checkin: mock.Mock + mock_capture_checkin: mock.Mock, + sentry_init: Callable[..., None], + span_streaming: bool, ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1496,9 +1560,13 @@ def test_child_process_record_checkin( ) -def test_child_process_pass_headers(sentry_init: Callable[..., None]) -> None: +@pytest.mark.parametrize("span_streaming", (False, True)) +def test_child_process_pass_headers(sentry_init: Callable[..., None], span_streaming: bool) -> None: """Task with pass_headers=True receives headers from the activation.""" - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1528,11 +1596,17 @@ def test_child_process_pass_headers(sentry_init: Callable[..., None]) -> None: redis.delete("task-headers-value", "task-headers-count", "task-headers-custom") +@pytest.mark.parametrize("span_streaming", (False, True)) @mock.patch("taskbroker_client.worker.workerchild.logger") def test_child_process_terminate_task( - sentry_init: Callable[..., None], mock_logger: mock.Mock + mock_logger: mock.Mock, + sentry_init: Callable[..., None], + span_streaming: bool, ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1579,11 +1653,17 @@ def test_child_process_terminate_task( assert "execution deadline" in extra["exception_message"] +@pytest.mark.parametrize("span_streaming", (False, True)) @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") def test_child_process_decompression( - sentry_init: Callable[..., None], mock_capture_checkin: mock.MagicMock + mock_capture_checkin: mock.MagicMock, + sentry_init: Callable[..., None], + span_streaming: bool, ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1609,9 +1689,15 @@ def test_child_process_decompression( assert mock_capture_checkin.call_count == 0 -def test_child_process_context_hooks(sentry_init: Callable[..., None]) -> None: +@pytest.mark.parametrize("span_streaming", (False, True)) +def test_child_process_context_hooks( + sentry_init: Callable[..., None], span_streaming: bool +) -> None: """Context hooks' on_execute is called with activation headers during task execution.""" - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) executed_headers: list[dict[str, str]] = [] @@ -1668,11 +1754,17 @@ def on_execute(self, headers: dict[str, str]) -> contextlib.AbstractContextManag app.context_hooks.remove(hook) +@pytest.mark.parametrize("span_streaming", (False, True)) @mock.patch("taskbroker_client.worker.workerchild.logger") def test_child_process_silenced_timeout( - sentry_init: Callable[..., None], mock_logger: mock.Mock + mock_logger: mock.Mock, + sentry_init: Callable[..., None], + span_streaming: bool, ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1703,11 +1795,17 @@ def test_child_process_silenced_timeout( assert failed_calls == [] +@pytest.mark.parametrize("span_streaming", (False, True)) @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") def test_child_process_silenced_exception_with_retries( - sentry_init: Callable[..., None], mock_capture: mock.Mock + mock_capture: mock.Mock, + sentry_init: Callable[..., None], + span_streaming: bool, ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1735,11 +1833,17 @@ def test_child_process_silenced_exception_with_retries( assert mock_capture.call_count == 0 +@pytest.mark.parametrize("span_streaming", (False, True)) @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") def test_child_process_expected_ignored_exception_max_attempts( - sentry_init: Callable[..., None], mock_capture: mock.Mock + mock_capture: mock.Mock, + sentry_init: Callable[..., None], + span_streaming: bool, ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1767,13 +1871,20 @@ def test_child_process_expected_ignored_exception_max_attempts( assert mock_capture.call_count == 0 +@pytest.mark.parametrize("span_streaming", (False, True)) @mock.patch("taskbroker_client.worker.workerchild.logger") @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") def test_child_process_silenced_exception_max_attempts( - sentry_init: Callable[..., None], mock_capture: mock.Mock, mock_logger: mock.Mock + mock_capture: mock.Mock, + mock_logger: mock.Mock, + sentry_init: Callable[..., None], + span_streaming: bool, ) -> None: """Silenced exceptions do not raise on retry exhaustion.""" - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) activation = InflightTaskActivation( host="localhost:50051", @@ -1825,11 +1936,17 @@ def test_child_process_silenced_exception_max_attempts( assert kwargs["extra"]["exception_type"] == "RuntimeError" +@pytest.mark.parametrize("span_streaming", (False, True)) @mock.patch("taskbroker_client.worker.workerchild.logger") def test_child_process_retry_on_deadline_exceeded( - sentry_init: Callable[..., None], mock_logger: mock.Mock + mock_logger: mock.Mock, + sentry_init: Callable[..., None], + span_streaming: bool, ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1862,12 +1979,18 @@ def test_child_process_retry_on_deadline_exceeded( assert kwargs["extra"]["exception_type"] == "ProcessingDeadlineExceeded" +@pytest.mark.parametrize("span_streaming", (False, True)) @mock.patch("taskbroker_client.worker.workerchild.logger") def test_child_process_general_exception_logs_task_failed( - sentry_init: Callable[..., None], mock_logger: mock.Mock + mock_logger: mock.Mock, + sentry_init: Callable[..., None], + span_streaming: bool, ) -> None: """A non-retriable Exception emits taskworker.task.failed with all fields.""" - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1902,14 +2025,19 @@ def test_child_process_general_exception_logs_task_failed( assert "exception_message" in extra +@pytest.mark.parametrize("span_streaming", (False, True)) @mock.patch("taskbroker_client.worker.workerchild.logger") def test_child_process_silenced_exception_does_not_log_task_failed( - sentry_init: Callable[..., None], mock_logger: mock.Mock, + sentry_init: Callable[..., None], + span_streaming: bool, ) -> None: """When err is in silenced_exceptions, taskworker.task.failed is NOT logged. Preserves the silencing semantics added in #608.""" - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -2003,14 +2131,19 @@ def _producing_task(task_id: str = "task-with-futures") -> InflightTaskActivatio ) +@pytest.mark.parametrize("span_streaming", (False, True)) @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_tracks_producer_futures( sentry_init: Callable[..., None], + span_streaming: bool, producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) task = _producing_task() todo: queue.Queue[InflightTaskActivation] = queue.Queue() @@ -2044,14 +2177,19 @@ def test_child_process_tracks_producer_futures( assert result.status == TASK_ACTIVATION_STATUS_COMPLETE +@pytest.mark.parametrize("span_streaming", (False, True)) @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_holds_result_until_futures_done( sentry_init: Callable[..., None], + span_streaming: bool, producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) task = _producing_task() todo: queue.Queue[InflightTaskActivation] = queue.Queue() @@ -2102,14 +2240,19 @@ def observe_and_resolve() -> None: assert result.status == TASK_ACTIVATION_STATUS_COMPLETE +@pytest.mark.parametrize("span_streaming", (False, True)) @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_skip_awaiting_futures_places_result_immediately( sentry_init: Callable[..., None], + span_streaming: bool, producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) task = _producing_task() todo: queue.Queue[InflightTaskActivation] = queue.Queue() @@ -2167,14 +2310,19 @@ def observe_and_resolve() -> None: assert processed.empty() +@pytest.mark.parametrize("span_streaming", (False, True)) @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_drains_pending_futures_on_sigterm( sentry_init: Callable[..., None], + span_streaming: bool, producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) task = _producing_task() todo: queue.Queue[InflightTaskActivation] = queue.Queue() @@ -2218,14 +2366,19 @@ def deliver_sigterm() -> None: assert result.status == TASK_ACTIVATION_STATUS_COMPLETE +@pytest.mark.parametrize("span_streaming", (False, True)) @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_retries_on_failed_future( sentry_init: Callable[..., None], + span_streaming: bool, producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) retriable_task = InflightTaskActivation( host="localhost:50051", @@ -2271,14 +2424,19 @@ def test_child_process_retries_on_failed_future( assert result.status == TASK_ACTIVATION_STATUS_RETRY +@pytest.mark.parametrize("span_streaming", (False, True)) @pytest.mark.parametrize("pending_registry", _PENDING_REGISTRIES) def test_child_process_clears_pending_futures_when_task_fails( sentry_init: Callable[..., None], + span_streaming: bool, pending_registry: Any, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) leftover_future: Future[BrokerValue[KafkaPayload]] = Future() leftover_future.set_result(_make_broker_value()) @@ -2312,11 +2470,18 @@ def test_child_process_clears_pending_futures_when_task_fails( assert len(pending_registry) == 0 +@pytest.mark.parametrize("span_streaming", (False, True)) def test_child_process_uses_configured_future_checking_frequency( - sentry_init: Callable[..., None], clear_pending_futures: None, restore_signal_handlers: None + sentry_init: Callable[..., None], + span_streaming: bool, + clear_pending_futures: None, + restore_signal_handlers: None, ) -> None: """The idle future-checking loop polls on the configured interval.""" - sentry_init(traces_sample_rate=1.0) + sentry_init( + traces_sample_rate=1.0, + _experiments={"trace_lifecycle": "stream" if span_streaming else "static"}, + ) # A task that runs long enough for the idle future-checking loop to poll a # few times before max_task_count triggers shutdown. From 49883c800345552eea6ab397abb624e24e07bfe1 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 14:03:50 +0200 Subject: [PATCH 09/30] move fixtures after mocks --- clients/python/tests/worker/test_worker.py | 39 ++++++++++++++-------- 1 file changed, 26 insertions(+), 13 deletions(-) diff --git a/clients/python/tests/worker/test_worker.py b/clients/python/tests/worker/test_worker.py index 5769228d..ad8677d3 100644 --- a/clients/python/tests/worker/test_worker.py +++ b/clients/python/tests/worker/test_worker.py @@ -1023,7 +1023,8 @@ def test_push_task_worker_busy(self) -> None: @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") def test_child_process_complete( - sentry_init: Callable[..., None], mock_capture_checkin: mock.MagicMock + mock_capture_checkin: mock.MagicMock, + sentry_init: Callable[..., None], ) -> None: sentry_init(traces_sample_rate=1.0) @@ -1116,8 +1117,8 @@ def test_child_process_emits_running_message(sentry_init: Callable[..., None]) - @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") def test_child_process_emits_exiting_once_and_continues_until_release( - sentry_init: Callable[..., None], mock_capture_checkin: mock.MagicMock, + sentry_init: Callable[..., None], ) -> None: sentry_init(traces_sample_rate=1.0) @@ -1282,7 +1283,9 @@ def test_child_process_retry_task(sentry_init: Callable[..., None]) -> None: @mock.patch("taskbroker_client.worker.workerchild.logger") @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") def test_child_process_retry_task_max_attempts( - sentry_init: Callable[..., None], mock_capture: mock.Mock, mock_logger: mock.Mock + mock_capture: mock.Mock, + mock_logger: mock.Mock, + sentry_init: Callable[..., None], ) -> None: sentry_init(traces_sample_rate=1.0) @@ -1461,7 +1464,8 @@ def test_child_process_at_most_once(sentry_init: Callable[..., None]) -> None: @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") def test_child_process_record_checkin( - sentry_init: Callable[..., None], mock_capture_checkin: mock.Mock + mock_capture_checkin: mock.Mock, + sentry_init: Callable[..., None], ) -> None: sentry_init(traces_sample_rate=1.0) @@ -1530,7 +1534,8 @@ def test_child_process_pass_headers(sentry_init: Callable[..., None]) -> None: @mock.patch("taskbroker_client.worker.workerchild.logger") def test_child_process_terminate_task( - sentry_init: Callable[..., None], mock_logger: mock.Mock + mock_logger: mock.Mock, + sentry_init: Callable[..., None], ) -> None: sentry_init(traces_sample_rate=1.0) @@ -1581,7 +1586,8 @@ def test_child_process_terminate_task( @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") def test_child_process_decompression( - sentry_init: Callable[..., None], mock_capture_checkin: mock.MagicMock + mock_capture_checkin: mock.MagicMock, + sentry_init: Callable[..., None], ) -> None: sentry_init(traces_sample_rate=1.0) @@ -1670,7 +1676,8 @@ def on_execute(self, headers: dict[str, str]) -> contextlib.AbstractContextManag @mock.patch("taskbroker_client.worker.workerchild.logger") def test_child_process_silenced_timeout( - sentry_init: Callable[..., None], mock_logger: mock.Mock + mock_logger: mock.Mock, + sentry_init: Callable[..., None], ) -> None: sentry_init(traces_sample_rate=1.0) @@ -1705,7 +1712,8 @@ def test_child_process_silenced_timeout( @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") def test_child_process_silenced_exception_with_retries( - sentry_init: Callable[..., None], mock_capture: mock.Mock + mock_capture: mock.Mock, + sentry_init: Callable[..., None], ) -> None: sentry_init(traces_sample_rate=1.0) @@ -1737,7 +1745,8 @@ def test_child_process_silenced_exception_with_retries( @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") def test_child_process_expected_ignored_exception_max_attempts( - sentry_init: Callable[..., None], mock_capture: mock.Mock + mock_capture: mock.Mock, + sentry_init: Callable[..., None], ) -> None: sentry_init(traces_sample_rate=1.0) @@ -1770,7 +1779,9 @@ def test_child_process_expected_ignored_exception_max_attempts( @mock.patch("taskbroker_client.worker.workerchild.logger") @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") def test_child_process_silenced_exception_max_attempts( - sentry_init: Callable[..., None], mock_capture: mock.Mock, mock_logger: mock.Mock + mock_capture: mock.Mock, + mock_logger: mock.Mock, + sentry_init: Callable[..., None], ) -> None: """Silenced exceptions do not raise on retry exhaustion.""" sentry_init(traces_sample_rate=1.0) @@ -1827,7 +1838,8 @@ def test_child_process_silenced_exception_max_attempts( @mock.patch("taskbroker_client.worker.workerchild.logger") def test_child_process_retry_on_deadline_exceeded( - sentry_init: Callable[..., None], mock_logger: mock.Mock + mock_logger: mock.Mock, + sentry_init: Callable[..., None], ) -> None: sentry_init(traces_sample_rate=1.0) @@ -1864,7 +1876,8 @@ def test_child_process_retry_on_deadline_exceeded( @mock.patch("taskbroker_client.worker.workerchild.logger") def test_child_process_general_exception_logs_task_failed( - sentry_init: Callable[..., None], mock_logger: mock.Mock + mock_logger: mock.Mock, + sentry_init: Callable[..., None], ) -> None: """A non-retriable Exception emits taskworker.task.failed with all fields.""" sentry_init(traces_sample_rate=1.0) @@ -1904,8 +1917,8 @@ def test_child_process_general_exception_logs_task_failed( @mock.patch("taskbroker_client.worker.workerchild.logger") def test_child_process_silenced_exception_does_not_log_task_failed( - sentry_init: Callable[..., None], mock_logger: mock.Mock, + sentry_init: Callable[..., None], ) -> None: """When err is in silenced_exceptions, taskworker.task.failed is NOT logged. Preserves the silencing semantics added in #608.""" From d6ea80956c3a6e45313cc4dbd2aacc3aef6e70a6 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 14:10:52 +0200 Subject: [PATCH 10/30] remove sentry_init from frequency test --- clients/python/tests/worker/test_worker.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/clients/python/tests/worker/test_worker.py b/clients/python/tests/worker/test_worker.py index ad8677d3..5e8ea8ea 100644 --- a/clients/python/tests/worker/test_worker.py +++ b/clients/python/tests/worker/test_worker.py @@ -2326,11 +2326,9 @@ def test_child_process_clears_pending_futures_when_task_fails( def test_child_process_uses_configured_future_checking_frequency( - sentry_init: Callable[..., None], clear_pending_futures: None, restore_signal_handlers: None + clear_pending_futures: None, restore_signal_handlers: None ) -> None: """The idle future-checking loop polls on the configured interval.""" - sentry_init(traces_sample_rate=1.0) - # A task that runs long enough for the idle future-checking loop to poll a # few times before max_task_count triggers shutdown. slow_task = InflightTaskActivation( From 9c64fa8567d907d6b05a279ed147b4373c7422d9 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 14:32:32 +0200 Subject: [PATCH 11/30] configure sdk to avoid sleeps --- clients/python/tests/worker/test_worker.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/clients/python/tests/worker/test_worker.py b/clients/python/tests/worker/test_worker.py index 5e8ea8ea..4fa0b5bc 100644 --- a/clients/python/tests/worker/test_worker.py +++ b/clients/python/tests/worker/test_worker.py @@ -2326,9 +2326,14 @@ def test_child_process_clears_pending_futures_when_task_fails( def test_child_process_uses_configured_future_checking_frequency( - clear_pending_futures: None, restore_signal_handlers: None + sentry_init: Callable[..., None], clear_pending_futures: None, restore_signal_handlers: None ) -> None: """The idle future-checking loop polls on the configured interval.""" + sentry_init( + traces_sample_rate=1.0, + enable_backpressure_handling=False, # To avoid time.sleep which the test patches. + ) + # A task that runs long enough for the idle future-checking loop to poll a # few times before max_task_count triggers shutdown. slow_task = InflightTaskActivation( From ea6dba32c9d737b7b96275486f8cf8fe460f8adc Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 14:54:35 +0200 Subject: [PATCH 12/30] fix import ordering problem --- clients/python/tests/worker/test_worker.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/clients/python/tests/worker/test_worker.py b/clients/python/tests/worker/test_worker.py index 4fa0b5bc..40b4f13d 100644 --- a/clients/python/tests/worker/test_worker.py +++ b/clients/python/tests/worker/test_worker.py @@ -2360,6 +2360,8 @@ def recording_sleep(seconds: float) -> None: idle_sleeps.append(seconds) real_sleep(seconds) + import examples.tasks # noqa: F401 — ensure sleep ref is bound before patching + # time.sleep is only used by the idle branch of check_task_future_completion # inside workerchild, so every recorded call comes from that loop. The task's # own sleep uses a separate `from time import sleep` import in examples.tasks. From 32fd19dcc7d8e60bd8c6bc6912f847b5394c4533 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 15:00:15 +0200 Subject: [PATCH 13/30] cleanup in fixture --- clients/python/tests/conftest.py | 3 +++ clients/python/tests/worker/test_worker.py | 2 +- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/clients/python/tests/conftest.py b/clients/python/tests/conftest.py index a4ad61a8..8c945d73 100644 --- a/clients/python/tests/conftest.py +++ b/clients/python/tests/conftest.py @@ -40,6 +40,9 @@ def inner(*a: Any, **kw: Any) -> None: sentry_sdk.get_current_scope().set_client(None) yield inner finally: + current = sentry_sdk.get_global_scope().client + if current is not None: + current.close() sentry_sdk.get_global_scope().set_client(old_client) diff --git a/clients/python/tests/worker/test_worker.py b/clients/python/tests/worker/test_worker.py index 40b4f13d..858ad939 100644 --- a/clients/python/tests/worker/test_worker.py +++ b/clients/python/tests/worker/test_worker.py @@ -2360,7 +2360,7 @@ def recording_sleep(seconds: float) -> None: idle_sleeps.append(seconds) real_sleep(seconds) - import examples.tasks # noqa: F401 — ensure sleep ref is bound before patching + import examples.tasks # noqa: F401; Ensure time.sleep reference is set before patching. # time.sleep is only used by the idle branch of check_task_future_completion # inside workerchild, so every recorded call comes from that loop. The task's From 265e1266df6aace2d591aa8bc207eebcfd8423fd Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 15:16:48 +0200 Subject: [PATCH 14/30] remove unused parameter --- clients/python/tests/conftest.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/clients/python/tests/conftest.py b/clients/python/tests/conftest.py index 8c945d73..db436d8e 100644 --- a/clients/python/tests/conftest.py +++ b/clients/python/tests/conftest.py @@ -6,7 +6,6 @@ import sentry_sdk import time_machine from arroyo.backends.kafka import KafkaProducer -from pytest import FixtureRequest from sentry_sdk.envelope import Envelope from sentry_sdk.transport import Transport @@ -29,7 +28,7 @@ def freeze_time(t: str | datetime | None = None) -> time_machine.travel: @pytest.fixture -def sentry_init(request: FixtureRequest) -> Generator[Callable[..., None], None, None]: +def sentry_init() -> Generator[Callable[..., None], None, None]: def inner(*a: Any, **kw: Any) -> None: kw.setdefault("transport", TestTransport()) client = sentry_sdk.Client(*a, **kw) From 2b52bbcfb0f38988b0d34f9a1357dc9751899bed Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 15:22:40 +0200 Subject: [PATCH 15/30] simplify transport --- clients/python/tests/conftest.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/clients/python/tests/conftest.py b/clients/python/tests/conftest.py index db436d8e..7877d095 100644 --- a/clients/python/tests/conftest.py +++ b/clients/python/tests/conftest.py @@ -46,9 +46,6 @@ def inner(*a: Any, **kw: Any) -> None: class TestTransport(Transport): - def __init__(self) -> None: - Transport.__init__(self) - def capture_envelope(self, _: Envelope) -> None: """No-op capture_envelope for tests""" pass From 71deb342154c8b296b625209aeb38261fbfd8a6d Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 15:26:20 +0200 Subject: [PATCH 16/30] add attributes in streaming path --- clients/python/src/taskbroker_client/worker/workerchild.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py index 6d05221c..f282750c 100644 --- a/clients/python/src/taskbroker_client/worker/workerchild.py +++ b/clients/python/src/taskbroker_client/worker/workerchild.py @@ -655,6 +655,9 @@ def _execute_activation( attributes={ "sentry.op": "queue.task.taskworker", "sentry.origin": "taskworker", + "args": args, + "kwargs": kwargs, + "id": activation.id, }, ) else: From bf93e3cd1cce1cdc834f1c9b6f58de08fbaf55b9 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 15:33:15 +0200 Subject: [PATCH 17/30] use consistent argument style --- clients/python/src/taskbroker_client/worker/workerchild.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py index f282750c..50a34b56 100644 --- a/clients/python/src/taskbroker_client/worker/workerchild.py +++ b/clients/python/src/taskbroker_client/worker/workerchild.py @@ -691,7 +691,7 @@ def _execute_activation( child_span: AbstractContextManager[Any] = ( sentry_sdk.traces.start_span( - activation.taskname, + name=activation.taskname, attributes={ "sentry.op": OP.QUEUE_PROCESS, "sentry.origin": "taskworker", From 084b3a19e9bc0df1269d1bc97c6cbe03faa4696f Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 15:47:33 +0200 Subject: [PATCH 18/30] create span only in isolation scope --- .../taskbroker_client/worker/workerchild.py | 162 +++++++++--------- 1 file changed, 81 insertions(+), 81 deletions(-) diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py index 50a34b56..52b01ac6 100644 --- a/clients/python/src/taskbroker_client/worker/workerchild.py +++ b/clients/python/src/taskbroker_client/worker/workerchild.py @@ -636,103 +636,103 @@ def _execute_activation( args = parameters.get("args", []) kwargs = parameters.get("kwargs", {}) - is_span_streaming = has_span_streaming_enabled(sentry_sdk.get_client().options) - - headers = dict(activation.headers) - sampling_context = { - "taskworker": { - "task": activation.taskname, - } - } - - parent_span: Transaction | NoOpSpan | StreamedSpan - if is_span_streaming: - sentry_sdk.traces.continue_trace(headers) - Scope.set_custom_sampling_context(sampling_context) - - parent_span = sentry_sdk.traces.start_span( - name=activation.taskname, - attributes={ - "sentry.op": "queue.task.taskworker", - "sentry.origin": "taskworker", - "args": args, - "kwargs": kwargs, - "id": activation.id, - }, - ) - else: - transaction = sentry_sdk.continue_trace( - environ_or_headers=headers, - op="queue.task.taskworker", - name=activation.taskname, - origin="taskworker", - ) - - parent_span = sentry_sdk.start_transaction( - transaction, custom_sampling_context=sampling_context - ) - with ( metrics.track_memory_usage( "taskworker.worker.memory_change", tags={"namespace": activation.namespace, "taskname": activation.taskname}, ), sentry_sdk.isolation_scope(), - parent_span, ): - if isinstance(parent_span, Span): - parent_span.set_data( - "taskworker-task", {"args": args, "kwargs": kwargs, "id": activation.id} - ) + is_span_streaming = has_span_streaming_enabled(sentry_sdk.get_client().options) + + headers = dict(activation.headers) + sampling_context = { + "taskworker": { + "task": activation.taskname, + } + } - task_added_time = activation.received_at.ToDatetime().timestamp() - # latency attribute needs to be in milliseconds - latency = (time.time() - task_added_time) * 1000 + parent_span: Transaction | NoOpSpan | StreamedSpan + if is_span_streaming: + sentry_sdk.traces.continue_trace(headers) + Scope.set_custom_sampling_context(sampling_context) - child_span: AbstractContextManager[Any] = ( - sentry_sdk.traces.start_span( + parent_span = sentry_sdk.traces.start_span( name=activation.taskname, attributes={ - "sentry.op": OP.QUEUE_PROCESS, + "sentry.op": "queue.task.taskworker", "sentry.origin": "taskworker", - SPANDATA.MESSAGING_DESTINATION_NAME: activation.namespace, - SPANDATA.MESSAGING_MESSAGE_ID: activation.id, - SPANDATA.MESSAGING_MESSAGE_RECEIVE_LATENCY: latency, - SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT: activation.retry_state.attempts, - SPANDATA.MESSAGING_SYSTEM: "taskworker", + "args": args, + "kwargs": kwargs, + "id": activation.id, }, ) - if is_span_streaming - else _task_processing_span(activation=activation, latency=latency) - ) + else: + transaction = sentry_sdk.continue_trace( + environ_or_headers=headers, + op="queue.task.taskworker", + name=activation.taskname, + origin="taskworker", + ) - with child_span: - # TODO(taskworker) remove this when doing cleanup - # The `__start_time` parameter is spliced into task parameters by - # sentry.celery.SentryTask._add_metadata and needs to be removed - # from kwargs like sentry.tasks.base.instrumented_task does. - if "__start_time" in kwargs: - kwargs.pop("__start_time") - - with contextlib.ExitStack() as stack: - with metrics.timer( - "taskworker.worker.context_rebuild.duration", - tags={ - "namespace": activation.namespace, - "taskname": activation.taskname, + parent_span = sentry_sdk.start_transaction( + transaction, custom_sampling_context=sampling_context + ) + + with parent_span: + if isinstance(parent_span, Span): + parent_span.set_data( + "taskworker-task", {"args": args, "kwargs": kwargs, "id": activation.id} + ) + + task_added_time = activation.received_at.ToDatetime().timestamp() + # latency attribute needs to be in milliseconds + latency = (time.time() - task_added_time) * 1000 + + child_span: AbstractContextManager[Any] = ( + sentry_sdk.traces.start_span( + name=activation.taskname, + attributes={ + "sentry.op": OP.QUEUE_PROCESS, + "sentry.origin": "taskworker", + SPANDATA.MESSAGING_DESTINATION_NAME: activation.namespace, + SPANDATA.MESSAGING_MESSAGE_ID: activation.id, + SPANDATA.MESSAGING_MESSAGE_RECEIVE_LATENCY: latency, + SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT: activation.retry_state.attempts, + SPANDATA.MESSAGING_SYSTEM: "taskworker", }, - ): - for hook in context_hooks: - stack.enter_context(hook.on_execute(headers)) - if task_func.pass_headers: - if "headers" in kwargs: - raise TypeError( - f"Task '{task_func.name}' has pass_headers=True, but 'headers' was passed in kwargs. " - "The 'headers' parameter is injected by the worker and cannot be passed by the caller." - ) - task_func(*args, headers=headers, **kwargs) - else: - task_func(*args, **kwargs) + ) + if is_span_streaming + else _task_processing_span(activation=activation, latency=latency) + ) + + with child_span: + # TODO(taskworker) remove this when doing cleanup + # The `__start_time` parameter is spliced into task parameters by + # sentry.celery.SentryTask._add_metadata and needs to be removed + # from kwargs like sentry.tasks.base.instrumented_task does. + if "__start_time" in kwargs: + kwargs.pop("__start_time") + + with contextlib.ExitStack() as stack: + with metrics.timer( + "taskworker.worker.context_rebuild.duration", + tags={ + "namespace": activation.namespace, + "taskname": activation.taskname, + }, + ): + for hook in context_hooks: + stack.enter_context(hook.on_execute(headers)) + if task_func.pass_headers: + if "headers" in kwargs: + raise TypeError( + f"Task '{task_func.name}' has pass_headers=True, but 'headers' was passed in kwargs. " + "The 'headers' parameter is injected by the worker and cannot be passed by the caller." + ) + task_func(*args, headers=headers, **kwargs) + else: + task_func(*args, **kwargs) def record_task_execution( activation: TaskActivation, From aad2d46f1820cc2d24e23bd84ccdea96a5734fc1 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Tue, 21 Jul 2026 15:51:12 +0200 Subject: [PATCH 19/30] add namespace to attributes --- clients/python/src/taskbroker_client/worker/workerchild.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py index 52b01ac6..1928e231 100644 --- a/clients/python/src/taskbroker_client/worker/workerchild.py +++ b/clients/python/src/taskbroker_client/worker/workerchild.py @@ -662,9 +662,9 @@ def _execute_activation( attributes={ "sentry.op": "queue.task.taskworker", "sentry.origin": "taskworker", - "args": args, - "kwargs": kwargs, - "id": activation.id, + "taskworker-task.args": args, + "taskworker-task.kwargs": kwargs, + "taskworker-task.id": activation.id, }, ) else: From 530308ada7ae3db32e497ddf40390ee4caa8db70 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Wed, 22 Jul 2026 13:38:10 +0200 Subject: [PATCH 20/30] Equivalent changes in taskbroker_client --- .../python/src/taskbroker_client/registry.py | 37 ++++++++++++++++--- 1 file changed, 31 insertions(+), 6 deletions(-) diff --git a/clients/python/src/taskbroker_client/registry.py b/clients/python/src/taskbroker_client/registry.py index 63e8ae4e..65cfc957 100644 --- a/clients/python/src/taskbroker_client/registry.py +++ b/clients/python/src/taskbroker_client/registry.py @@ -2,8 +2,9 @@ import datetime import logging -from collections.abc import Callable +from collections.abc import Callable, Generator from concurrent import futures +from contextlib import AbstractContextManager, contextmanager from typing import Any, cast import sentry_sdk @@ -11,6 +12,8 @@ from arroyo.types import BrokerValue, Topic from sentry_protos.taskbroker.v1.taskbroker_pb2 import TaskActivation from sentry_sdk.consts import OP, SPANDATA +from sentry_sdk.tracing import Span +from sentry_sdk.tracing_utils import has_span_streaming_enabled from taskbroker_client.constants import ( DEFAULT_PROCESSING_DEADLINE, @@ -167,11 +170,9 @@ def _handle_produce_future(self, future: ProducerFuture, tags: dict[str, str]) - else: self.metrics.incr("taskworker.registry.send_task.success", tags=tags) - def send_task( - self, activation: TaskActivation, wait_for_delivery: bool = False - ) -> ProducerFuture: - topic = self.topic - + @contextmanager + def _task_publishing_span(self, activation: TaskActivation) -> Generator[Span, None, None]: + """Provide a span in the transaction-based tracing API with relevant attributes set.""" with sentry_sdk.start_span( op=OP.QUEUE_PUBLISH, name=activation.taskname, @@ -180,7 +181,31 @@ def send_task( span.set_data(SPANDATA.MESSAGING_DESTINATION_NAME, activation.namespace) span.set_data(SPANDATA.MESSAGING_MESSAGE_ID, activation.id) span.set_data(SPANDATA.MESSAGING_SYSTEM, "taskworker") + yield span + + def send_task( + self, activation: TaskActivation, wait_for_delivery: bool = False + ) -> ProducerFuture: + topic = self.topic + + is_span_streaming = has_span_streaming_enabled(sentry_sdk.get_client().options) + + span: AbstractContextManager[Any] = ( + sentry_sdk.traces.start_span( + name=activation.taskname, + attributes={ + "sentry.op": OP.QUEUE_PUBLISH, + "sentry.origin": "taskworker", + SPANDATA.MESSAGING_DESTINATION_NAME: activation.namespace, + SPANDATA.MESSAGING_MESSAGE_ID: activation.id, + SPANDATA.MESSAGING_SYSTEM: "taskworker", + }, + ) + if is_span_streaming + else self._task_publishing_span(activation=activation) + ) + with span: produce_future = self._producer(topic).produce( Topic(name=topic), KafkaPayload(key=None, value=activation.SerializeToString(), headers=[]), From ce10118377809f5c44405ed290838d715c25c20a Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Wed, 22 Jul 2026 14:05:10 +0200 Subject: [PATCH 21/30] update tests --- clients/python/tests/test_task.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/clients/python/tests/test_task.py b/clients/python/tests/test_task.py index ec5baa11..1ddf9ea9 100644 --- a/clients/python/tests/test_task.py +++ b/clients/python/tests/test_task.py @@ -310,7 +310,7 @@ def test_create_activation_tracing( 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 @@ -331,7 +331,7 @@ def test_create_activation_tracing_headers( 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"} ) @@ -355,7 +355,7 @@ def test_create_activation_tracing_disable( 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} ) From 35831f8eee61b7e909b57d95f37bd35cc74f0c11 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Wed, 22 Jul 2026 16:33:09 +0200 Subject: [PATCH 22/30] revert test changes --- clients/python/tests/conftest.py | 30 ---- clients/python/tests/test_task.py | 20 +-- clients/python/tests/worker/test_worker.py | 153 ++++----------------- 3 files changed, 27 insertions(+), 176 deletions(-) diff --git a/clients/python/tests/conftest.py b/clients/python/tests/conftest.py index 7877d095..2d777a25 100644 --- a/clients/python/tests/conftest.py +++ b/clients/python/tests/conftest.py @@ -1,13 +1,7 @@ -from collections.abc import Callable, Generator from datetime import UTC, datetime -from typing import Any -import pytest -import sentry_sdk import time_machine from arroyo.backends.kafka import KafkaProducer -from sentry_sdk.envelope import Envelope -from sentry_sdk.transport import Transport from taskbroker_client.types import AtMostOnceStore @@ -27,30 +21,6 @@ def freeze_time(t: str | datetime | None = None) -> time_machine.travel: return time_machine.travel(t, tick=False) -@pytest.fixture -def sentry_init() -> Generator[Callable[..., None], None, None]: - def inner(*a: Any, **kw: Any) -> None: - kw.setdefault("transport", TestTransport()) - client = sentry_sdk.Client(*a, **kw) - sentry_sdk.get_global_scope().set_client(client) - - old_client = sentry_sdk.get_global_scope().client - try: - sentry_sdk.get_current_scope().set_client(None) - yield inner - finally: - current = sentry_sdk.get_global_scope().client - if current is not None: - current.close() - sentry_sdk.get_global_scope().set_client(old_client) - - -class TestTransport(Transport): - def capture_envelope(self, _: Envelope) -> None: - """No-op capture_envelope for tests""" - pass - - class StubAtMostOnce(AtMostOnceStore): def __init__(self) -> None: self._keys: dict[str, str] = {} diff --git a/clients/python/tests/test_task.py b/clients/python/tests/test_task.py index c9e8501e..2556b9e7 100644 --- a/clients/python/tests/test_task.py +++ b/clients/python/tests/test_task.py @@ -2,7 +2,7 @@ import datetime from collections.abc import MutableMapping from concurrent.futures import Future -from typing import Any, Callable +from typing import Any from unittest.mock import patch import msgpack @@ -297,11 +297,7 @@ def with_parameters(one: str, two: int, org_id: int) -> None: assert activation.parameters == "" -def test_create_activation_tracing( - sentry_init: Callable[..., None], task_namespace: TaskNamespace -) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_create_activation_tracing(task_namespace: TaskNamespace) -> None: @task_namespace.register(name="test.parameters") def with_parameters(one: str, two: int, org_id: int) -> None: raise NotImplementedError @@ -314,11 +310,7 @@ def with_parameters(one: str, two: int, org_id: int) -> None: assert "baggage" in headers -def test_create_activation_tracing_headers( - sentry_init: Callable[..., None], task_namespace: TaskNamespace -) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_create_activation_tracing_headers(task_namespace: TaskNamespace) -> None: @task_namespace.register(name="test.parameters") def with_parameters(one: str, two: int, org_id: int) -> None: raise NotImplementedError @@ -334,11 +326,7 @@ def with_parameters(one: str, two: int, org_id: int) -> None: assert headers["key"] == "value" -def test_create_activation_tracing_disable( - sentry_init: Callable[..., None], task_namespace: TaskNamespace -) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_create_activation_tracing_disable(task_namespace: TaskNamespace) -> None: @task_namespace.register(name="test.parameters") def with_parameters(one: str, two: int, org_id: int) -> None: raise NotImplementedError diff --git a/clients/python/tests/worker/test_worker.py b/clients/python/tests/worker/test_worker.py index 858ad939..7253a74e 100644 --- a/clients/python/tests/worker/test_worker.py +++ b/clients/python/tests/worker/test_worker.py @@ -1022,12 +1022,7 @@ def test_push_task_worker_busy(self) -> None: @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") -def test_child_process_complete( - mock_capture_checkin: mock.MagicMock, - sentry_init: Callable[..., None], -) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_child_process_complete(mock_capture_checkin: mock.MagicMock) -> None: todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1052,11 +1047,7 @@ def test_child_process_complete( assert mock_capture_checkin.call_count == 0 -def test_child_process_canary_task( - sentry_init: Callable[..., None], capsys: pytest.CaptureFixture[str] -) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_child_process_canary_task(capsys: pytest.CaptureFixture[str]) -> None: todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1082,9 +1073,7 @@ def test_child_process_canary_task( assert capsys.readouterr().out == "Done running canary task!\n" -def test_child_process_emits_running_message(sentry_init: Callable[..., None]) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_child_process_emits_running_message() -> None: todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1118,10 +1107,7 @@ def test_child_process_emits_running_message(sentry_init: Callable[..., None]) - @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") def test_child_process_emits_exiting_once_and_continues_until_release( mock_capture_checkin: mock.MagicMock, - sentry_init: Callable[..., None], ) -> None: - sentry_init(traces_sample_rate=1.0) - shutdown = Event() ctx = get_context("fork") child_id = uuid4() @@ -1181,9 +1167,7 @@ def test_child_process_emits_exiting_once_and_continues_until_release( assert mock_capture_checkin.call_count == 0 -def test_child_process_emits_busy_and_idle_messages(sentry_init: Callable[..., None]) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_child_process_emits_busy_and_idle_messages() -> None: todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1215,9 +1199,7 @@ def test_child_process_emits_busy_and_idle_messages(sentry_init: Callable[..., N assert processed.get(timeout=1).task_id == SIMPLE_TASK.activation.id -def test_child_process_remove_start_time_kwargs(sentry_init: Callable[..., None]) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_child_process_remove_start_time_kwargs() -> None: activation = InflightTaskActivation( host="localhost:50051", receive_timestamp=0, @@ -1254,9 +1236,7 @@ def test_child_process_remove_start_time_kwargs(sentry_init: Callable[..., None] assert result.status == TASK_ACTIVATION_STATUS_COMPLETE -def test_child_process_retry_task(sentry_init: Callable[..., None]) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_child_process_retry_task() -> None: todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1283,12 +1263,8 @@ def test_child_process_retry_task(sentry_init: Callable[..., None]) -> None: @mock.patch("taskbroker_client.worker.workerchild.logger") @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") def test_child_process_retry_task_max_attempts( - mock_capture: mock.Mock, - mock_logger: mock.Mock, - sentry_init: Callable[..., None], + mock_capture: mock.Mock, mock_logger: mock.Mock ) -> None: - sentry_init(traces_sample_rate=1.0) - # Create an activation that is on its final attempt and # will raise an error again. activation = InflightTaskActivation( @@ -1348,9 +1324,7 @@ def test_child_process_retry_task_max_attempts( assert extra["retry_max_attempts"] == 3 -def test_child_process_failure_task(sentry_init: Callable[..., None]) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_child_process_failure_task() -> None: todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1374,9 +1348,7 @@ def test_child_process_failure_task(sentry_init: Callable[..., None]) -> None: assert result.status == TASK_ACTIVATION_STATUS_FAILURE -def test_child_process_shutdown(sentry_init: Callable[..., None]) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_child_process_shutdown() -> None: todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1400,9 +1372,7 @@ def test_child_process_shutdown(sentry_init: Callable[..., None]) -> None: assert processed.qsize() == 0 -def test_child_process_unknown_task(sentry_init: Callable[..., None]) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_child_process_unknown_task() -> None: todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1430,9 +1400,7 @@ def test_child_process_unknown_task(sentry_init: Callable[..., None]) -> None: assert result.status == TASK_ACTIVATION_STATUS_COMPLETE -def test_child_process_at_most_once(sentry_init: Callable[..., None]) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_child_process_at_most_once() -> None: todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1463,12 +1431,7 @@ def test_child_process_at_most_once(sentry_init: Callable[..., None]) -> None: @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") -def test_child_process_record_checkin( - mock_capture_checkin: mock.Mock, - sentry_init: Callable[..., None], -) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_child_process_record_checkin(mock_capture_checkin: mock.Mock) -> None: todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1500,10 +1463,8 @@ def test_child_process_record_checkin( ) -def test_child_process_pass_headers(sentry_init: Callable[..., None]) -> None: +def test_child_process_pass_headers() -> None: """Task with pass_headers=True receives headers from the activation.""" - sentry_init(traces_sample_rate=1.0) - todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1533,12 +1494,7 @@ def test_child_process_pass_headers(sentry_init: Callable[..., None]) -> None: @mock.patch("taskbroker_client.worker.workerchild.logger") -def test_child_process_terminate_task( - mock_logger: mock.Mock, - sentry_init: Callable[..., None], -) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_child_process_terminate_task(mock_logger: mock.Mock) -> None: todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1585,11 +1541,7 @@ def test_child_process_terminate_task( @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") -def test_child_process_decompression( - mock_capture_checkin: mock.MagicMock, - sentry_init: Callable[..., None], -) -> None: - sentry_init(traces_sample_rate=1.0) +def test_child_process_decompression(mock_capture_checkin: mock.MagicMock) -> None: todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1615,10 +1567,8 @@ def test_child_process_decompression( assert mock_capture_checkin.call_count == 0 -def test_child_process_context_hooks(sentry_init: Callable[..., None]) -> None: +def test_child_process_context_hooks() -> None: """Context hooks' on_execute is called with activation headers during task execution.""" - sentry_init(traces_sample_rate=1.0) - executed_headers: list[dict[str, str]] = [] class RecordingHook: @@ -1675,12 +1625,7 @@ def on_execute(self, headers: dict[str, str]) -> contextlib.AbstractContextManag @mock.patch("taskbroker_client.worker.workerchild.logger") -def test_child_process_silenced_timeout( - mock_logger: mock.Mock, - sentry_init: Callable[..., None], -) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_child_process_silenced_timeout(mock_logger: mock.Mock) -> None: todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1711,12 +1656,7 @@ def test_child_process_silenced_timeout( @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") -def test_child_process_silenced_exception_with_retries( - mock_capture: mock.Mock, - sentry_init: Callable[..., None], -) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_child_process_silenced_exception_with_retries(mock_capture: mock.Mock) -> None: todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1744,12 +1684,7 @@ def test_child_process_silenced_exception_with_retries( @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") -def test_child_process_expected_ignored_exception_max_attempts( - mock_capture: mock.Mock, - sentry_init: Callable[..., None], -) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_child_process_expected_ignored_exception_max_attempts(mock_capture: mock.Mock) -> None: todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1779,13 +1714,9 @@ def test_child_process_expected_ignored_exception_max_attempts( @mock.patch("taskbroker_client.worker.workerchild.logger") @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") def test_child_process_silenced_exception_max_attempts( - mock_capture: mock.Mock, - mock_logger: mock.Mock, - sentry_init: Callable[..., None], + mock_capture: mock.Mock, mock_logger: mock.Mock ) -> None: """Silenced exceptions do not raise on retry exhaustion.""" - sentry_init(traces_sample_rate=1.0) - activation = InflightTaskActivation( host="localhost:50051", receive_timestamp=0, @@ -1837,12 +1768,7 @@ def test_child_process_silenced_exception_max_attempts( @mock.patch("taskbroker_client.worker.workerchild.logger") -def test_child_process_retry_on_deadline_exceeded( - mock_logger: mock.Mock, - sentry_init: Callable[..., None], -) -> None: - sentry_init(traces_sample_rate=1.0) - +def test_child_process_retry_on_deadline_exceeded(mock_logger: mock.Mock) -> None: todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1875,13 +1801,8 @@ def test_child_process_retry_on_deadline_exceeded( @mock.patch("taskbroker_client.worker.workerchild.logger") -def test_child_process_general_exception_logs_task_failed( - mock_logger: mock.Mock, - sentry_init: Callable[..., None], -) -> None: +def test_child_process_general_exception_logs_task_failed(mock_logger: mock.Mock) -> None: """A non-retriable Exception emits taskworker.task.failed with all fields.""" - sentry_init(traces_sample_rate=1.0) - todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1918,12 +1839,9 @@ def test_child_process_general_exception_logs_task_failed( @mock.patch("taskbroker_client.worker.workerchild.logger") def test_child_process_silenced_exception_does_not_log_task_failed( mock_logger: mock.Mock, - sentry_init: Callable[..., None], ) -> None: """When err is in silenced_exceptions, taskworker.task.failed is NOT logged. Preserves the silencing semantics added in #608.""" - sentry_init(traces_sample_rate=1.0) - todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -2018,13 +1936,10 @@ def _producing_task(task_id: str = "task-with-futures") -> InflightTaskActivatio @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_tracks_producer_futures( - sentry_init: Callable[..., None], producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: - sentry_init(traces_sample_rate=1.0) - task = _producing_task() todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -2059,13 +1974,10 @@ def test_child_process_tracks_producer_futures( @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_holds_result_until_futures_done( - sentry_init: Callable[..., None], producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: - sentry_init(traces_sample_rate=1.0) - task = _producing_task() todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -2117,13 +2029,10 @@ def observe_and_resolve() -> None: @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_skip_awaiting_futures_places_result_immediately( - sentry_init: Callable[..., None], producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: - sentry_init(traces_sample_rate=1.0) - task = _producing_task() todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -2182,13 +2091,10 @@ def observe_and_resolve() -> None: @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_drains_pending_futures_on_sigterm( - sentry_init: Callable[..., None], producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: - sentry_init(traces_sample_rate=1.0) - task = _producing_task() todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -2233,13 +2139,10 @@ def deliver_sigterm() -> None: @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_retries_on_failed_future( - sentry_init: Callable[..., None], producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: - sentry_init(traces_sample_rate=1.0) - retriable_task = InflightTaskActivation( host="localhost:50051", receive_timestamp=0, @@ -2286,13 +2189,10 @@ def test_child_process_retries_on_failed_future( @pytest.mark.parametrize("pending_registry", _PENDING_REGISTRIES) def test_child_process_clears_pending_futures_when_task_fails( - sentry_init: Callable[..., None], pending_registry: Any, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: - sentry_init(traces_sample_rate=1.0) - leftover_future: Future[BrokerValue[KafkaPayload]] = Future() leftover_future.set_result(_make_broker_value()) pending_registry["test.producer"].append(leftover_future) @@ -2326,14 +2226,9 @@ def test_child_process_clears_pending_futures_when_task_fails( def test_child_process_uses_configured_future_checking_frequency( - sentry_init: Callable[..., None], clear_pending_futures: None, restore_signal_handlers: None + clear_pending_futures: None, restore_signal_handlers: None ) -> None: """The idle future-checking loop polls on the configured interval.""" - sentry_init( - traces_sample_rate=1.0, - enable_backpressure_handling=False, # To avoid time.sleep which the test patches. - ) - # A task that runs long enough for the idle future-checking loop to poll a # few times before max_task_count triggers shutdown. slow_task = InflightTaskActivation( @@ -2360,8 +2255,6 @@ def recording_sleep(seconds: float) -> None: idle_sleeps.append(seconds) real_sleep(seconds) - import examples.tasks # noqa: F401; Ensure time.sleep reference is set before patching. - # time.sleep is only used by the idle branch of check_task_future_completion # inside workerchild, so every recorded call comes from that loop. The task's # own sleep uses a separate `from time import sleep` import in examples.tasks. From d2cae9c995ef3e472684e996000a8ebff6c359de Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Wed, 22 Jul 2026 17:21:49 +0200 Subject: [PATCH 23/30] create shim in sdk.py --- .../python/src/taskbroker_client/registry.py | 51 ++--- clients/python/src/taskbroker_client/sdk.py | 69 +++++++ .../taskbroker_client/worker/workerchild.py | 179 +++++++----------- 3 files changed, 155 insertions(+), 144 deletions(-) create mode 100644 clients/python/src/taskbroker_client/sdk.py diff --git a/clients/python/src/taskbroker_client/registry.py b/clients/python/src/taskbroker_client/registry.py index 65cfc957..be281108 100644 --- a/clients/python/src/taskbroker_client/registry.py +++ b/clients/python/src/taskbroker_client/registry.py @@ -2,18 +2,15 @@ import datetime import logging -from collections.abc import Callable, Generator +from collections.abc import Callable from concurrent import futures -from contextlib import AbstractContextManager, contextmanager from typing import Any, cast -import sentry_sdk from arroyo.backends.kafka import KafkaPayload from arroyo.types import BrokerValue, Topic from sentry_protos.taskbroker.v1.taskbroker_pb2 import TaskActivation from sentry_sdk.consts import OP, SPANDATA -from sentry_sdk.tracing import Span -from sentry_sdk.tracing_utils import has_span_streaming_enabled +from sentry_sdk.traces import StreamedSpan from taskbroker_client.constants import ( DEFAULT_PROCESSING_DEADLINE, @@ -23,6 +20,7 @@ from taskbroker_client.metrics import MetricsBackend from taskbroker_client.retry import Retry from taskbroker_client.router import TaskRouter +from taskbroker_client.sdk import start_span from taskbroker_client.task import ExternalTask, P, R, Task from taskbroker_client.types import ContextHook, ProducerFactory, ProducerProtocol @@ -170,42 +168,25 @@ def _handle_produce_future(self, future: ProducerFuture, tags: dict[str, str]) - else: self.metrics.incr("taskworker.registry.send_task.success", tags=tags) - @contextmanager - def _task_publishing_span(self, activation: TaskActivation) -> Generator[Span, None, None]: - """Provide a span in the transaction-based tracing API with relevant attributes set.""" - with sentry_sdk.start_span( - op=OP.QUEUE_PUBLISH, - name=activation.taskname, - origin="taskworker", - ) as span: - span.set_data(SPANDATA.MESSAGING_DESTINATION_NAME, activation.namespace) - span.set_data(SPANDATA.MESSAGING_MESSAGE_ID, activation.id) - span.set_data(SPANDATA.MESSAGING_SYSTEM, "taskworker") - yield span - def send_task( self, activation: TaskActivation, wait_for_delivery: bool = False ) -> ProducerFuture: topic = self.topic - is_span_streaming = has_span_streaming_enabled(sentry_sdk.get_client().options) - - span: AbstractContextManager[Any] = ( - sentry_sdk.traces.start_span( - name=activation.taskname, - attributes={ - "sentry.op": OP.QUEUE_PUBLISH, - "sentry.origin": "taskworker", - SPANDATA.MESSAGING_DESTINATION_NAME: activation.namespace, - SPANDATA.MESSAGING_MESSAGE_ID: activation.id, - SPANDATA.MESSAGING_SYSTEM: "taskworker", - }, - ) - if is_span_streaming - else self._task_publishing_span(activation=activation) - ) + with start_span( + name=activation.taskname, + op=OP.QUEUE_PUBLISH, + origin="taskworker", + ) as span: + if isinstance(span, StreamedSpan): + span.set_attribute(SPANDATA.MESSAGING_DESTINATION_NAME, activation.namespace) + span.set_attribute(SPANDATA.MESSAGING_MESSAGE_ID, activation.id) + span.set_attribute(SPANDATA.MESSAGING_SYSTEM, "taskworker") + else: + span.set_data(SPANDATA.MESSAGING_DESTINATION_NAME, activation.namespace) + span.set_data(SPANDATA.MESSAGING_MESSAGE_ID, activation.id) + span.set_data(SPANDATA.MESSAGING_SYSTEM, "taskworker") - with span: produce_future = self._producer(topic).produce( Topic(name=topic), KafkaPayload(key=None, value=activation.SerializeToString(), headers=[]), diff --git a/clients/python/src/taskbroker_client/sdk.py b/clients/python/src/taskbroker_client/sdk.py new file mode 100644 index 00000000..e9071923 --- /dev/null +++ b/clients/python/src/taskbroker_client/sdk.py @@ -0,0 +1,69 @@ +from contextlib import nullcontext +from typing import Any, ContextManager + +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( + name: str, origin: str, 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 + 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) + + return sentry_sdk.traces.start_span( + name=name, + attributes={ + "sentry.origin": origin, + }, + ) + + transaction = sentry_sdk.continue_trace( + environ_or_headers=headers, + op="queue.task.taskworker", + name=name, + origin=origin, + ) + + span = sentry_sdk.start_transaction(transaction, custom_sampling_context=sampling_context) + except Exception: + pass + + if span is None: + return nullcontext() + return span + + +def start_span(name: str, op: str, origin: str) -> Span | StreamedSpan | ContextManager[Any]: + """Start a span in the currently active trace lifecycle.""" + span = None + try: + is_span_streaming = has_span_streaming_enabled(sentry_sdk.get_client().options) + if is_span_streaming: + return sentry_sdk.traces.start_span( + name=name, + attributes={ + "sentry.op": op, + "sentry.origin": origin, + }, + ) + + span = sentry_sdk.start_span( + op=op, + name=name, + origin=origin, + ) + except Exception: + pass + + if span is None: + return nullcontext() + return span diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py index 1928e231..d3ee9bb4 100644 --- a/clients/python/src/taskbroker_client/worker/workerchild.py +++ b/clients/python/src/taskbroker_client/worker/workerchild.py @@ -8,7 +8,6 @@ import threading import time from collections.abc import Callable, Generator, Sequence -from contextlib import AbstractContextManager, contextmanager from dataclasses import dataclass from functools import partial from multiprocessing.synchronize import Event @@ -33,14 +32,13 @@ ) from sentry_sdk.consts import OP, SPANDATA from sentry_sdk.crons import MonitorStatus, capture_checkin -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 +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.state import clear_current_task, current_task, set_current_task from taskbroker_client.task import Task from taskbroker_client.types import ContextHook, InflightTaskActivation, ProcessingResult @@ -605,26 +603,6 @@ def check_task_future_completion( for task in pending_task_futures.copy(): await_task_futures(task) - @contextmanager - def _task_processing_span( - activation: TaskActivation, latency: float - ) -> Generator[Span, None, None]: - """Provide a span in the transaction-based tracing API with relevant attributes set.""" - with sentry_sdk.start_span( - op=OP.QUEUE_PROCESS, - name=activation.taskname, - origin="taskworker", - ) as span: - span.set_data(SPANDATA.MESSAGING_DESTINATION_NAME, activation.namespace) - span.set_data(SPANDATA.MESSAGING_MESSAGE_ID, activation.id) - span.set_data(SPANDATA.MESSAGING_MESSAGE_RECEIVE_LATENCY, latency) - span.set_data( - SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT, - activation.retry_state.attempts, - ) - span.set_data(SPANDATA.MESSAGING_SYSTEM, "taskworker") - yield span - def _execute_activation( task_func: Task[Any, Any], activation: TaskActivation, @@ -636,103 +614,86 @@ def _execute_activation( args = parameters.get("args", []) kwargs = parameters.get("kwargs", {}) + headers = dict(activation.headers) + with ( metrics.track_memory_usage( "taskworker.worker.memory_change", tags={"namespace": activation.namespace, "taskname": activation.taskname}, ), sentry_sdk.isolation_scope(), + start_transaction( + name=activation.taskname, + origin="taskworker", + headers=headers, + sampling_context={ + "taskworker": { + "task": activation.taskname, + } + }, + ) as parent_span, ): - is_span_streaming = has_span_streaming_enabled(sentry_sdk.get_client().options) - - headers = dict(activation.headers) - sampling_context = { - "taskworker": { - "task": activation.taskname, - } - } - - parent_span: Transaction | NoOpSpan | StreamedSpan - if is_span_streaming: - sentry_sdk.traces.continue_trace(headers) - Scope.set_custom_sampling_context(sampling_context) - - parent_span = sentry_sdk.traces.start_span( - name=activation.taskname, - attributes={ - "sentry.op": "queue.task.taskworker", - "sentry.origin": "taskworker", - "taskworker-task.args": args, - "taskworker-task.kwargs": kwargs, - "taskworker-task.id": activation.id, - }, - ) - else: - transaction = sentry_sdk.continue_trace( - environ_or_headers=headers, - op="queue.task.taskworker", - name=activation.taskname, - origin="taskworker", + if isinstance(parent_span, StreamedSpan): + parent_span.set_attribute("taskworker-task.args", args) + parent_span.set_attribute("taskworker-task.kwargs", kwargs) + parent_span.set_attribute("taskworker-task.id", activation.id) + elif isinstance(parent_span, Span): + parent_span.set_data( + "taskworker-task", {"args": args, "kwargs": kwargs, "id": activation.id} ) - parent_span = sentry_sdk.start_transaction( - transaction, custom_sampling_context=sampling_context - ) + task_added_time = activation.received_at.ToDatetime().timestamp() + # latency attribute needs to be in milliseconds + latency = (time.time() - task_added_time) * 1000 - with parent_span: - if isinstance(parent_span, Span): - parent_span.set_data( - "taskworker-task", {"args": args, "kwargs": kwargs, "id": activation.id} + with start_span( + name=activation.taskname, op=OP.QUEUE_PROCESS, origin="taskworker" + ) as child_span: + if isinstance(child_span, StreamedSpan): + child_span.set_attribute( + SPANDATA.MESSAGING_DESTINATION_NAME, activation.namespace ) - - task_added_time = activation.received_at.ToDatetime().timestamp() - # latency attribute needs to be in milliseconds - latency = (time.time() - task_added_time) * 1000 - - child_span: AbstractContextManager[Any] = ( - sentry_sdk.traces.start_span( - name=activation.taskname, - attributes={ - "sentry.op": OP.QUEUE_PROCESS, - "sentry.origin": "taskworker", - SPANDATA.MESSAGING_DESTINATION_NAME: activation.namespace, - SPANDATA.MESSAGING_MESSAGE_ID: activation.id, - SPANDATA.MESSAGING_MESSAGE_RECEIVE_LATENCY: latency, - SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT: activation.retry_state.attempts, - SPANDATA.MESSAGING_SYSTEM: "taskworker", - }, + child_span.set_attribute(SPANDATA.MESSAGING_MESSAGE_ID, activation.id) + child_span.set_attribute(SPANDATA.MESSAGING_MESSAGE_RECEIVE_LATENCY, latency) + child_span.set_attribute( + SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT, activation.retry_state.attempts ) - if is_span_streaming - else _task_processing_span(activation=activation, latency=latency) - ) - - with child_span: - # TODO(taskworker) remove this when doing cleanup - # The `__start_time` parameter is spliced into task parameters by - # sentry.celery.SentryTask._add_metadata and needs to be removed - # from kwargs like sentry.tasks.base.instrumented_task does. - if "__start_time" in kwargs: - kwargs.pop("__start_time") - - with contextlib.ExitStack() as stack: - with metrics.timer( - "taskworker.worker.context_rebuild.duration", - tags={ - "namespace": activation.namespace, - "taskname": activation.taskname, - }, - ): - for hook in context_hooks: - stack.enter_context(hook.on_execute(headers)) - if task_func.pass_headers: - if "headers" in kwargs: - raise TypeError( - f"Task '{task_func.name}' has pass_headers=True, but 'headers' was passed in kwargs. " - "The 'headers' parameter is injected by the worker and cannot be passed by the caller." - ) - task_func(*args, headers=headers, **kwargs) - else: - task_func(*args, **kwargs) + child_span.set_attribute(SPANDATA.MESSAGING_SYSTEM, "taskworker") + elif isinstance(parent_span, Span): + child_span.set_data(SPANDATA.MESSAGING_DESTINATION_NAME, activation.namespace) + child_span.set_data(SPANDATA.MESSAGING_MESSAGE_ID, activation.id) + child_span.set_data(SPANDATA.MESSAGING_MESSAGE_RECEIVE_LATENCY, latency) + child_span.set_data( + SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT, activation.retry_state.attempts + ) + child_span.set_data(SPANDATA.MESSAGING_SYSTEM, "taskworker") + + # TODO(taskworker) remove this when doing cleanup + # The `__start_time` parameter is spliced into task parameters by + # sentry.celery.SentryTask._add_metadata and needs to be removed + # from kwargs like sentry.tasks.base.instrumented_task does. + if "__start_time" in kwargs: + kwargs.pop("__start_time") + + with contextlib.ExitStack() as stack: + with metrics.timer( + "taskworker.worker.context_rebuild.duration", + tags={ + "namespace": activation.namespace, + "taskname": activation.taskname, + }, + ): + for hook in context_hooks: + stack.enter_context(hook.on_execute(headers)) + if task_func.pass_headers: + if "headers" in kwargs: + raise TypeError( + f"Task '{task_func.name}' has pass_headers=True, but 'headers' was passed in kwargs. " + "The 'headers' parameter is injected by the worker and cannot be passed by the caller." + ) + task_func(*args, headers=headers, **kwargs) + else: + task_func(*args, **kwargs) def record_task_execution( activation: TaskActivation, From aca495911a3f06dc6b4e2c63271218615c793f01 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Wed, 22 Jul 2026 17:28:52 +0200 Subject: [PATCH 24/30] address bots --- clients/python/src/taskbroker_client/registry.py | 3 ++- clients/python/src/taskbroker_client/sdk.py | 9 ++++++--- .../python/src/taskbroker_client/worker/workerchild.py | 3 ++- 3 files changed, 10 insertions(+), 5 deletions(-) diff --git a/clients/python/src/taskbroker_client/registry.py b/clients/python/src/taskbroker_client/registry.py index be281108..feff2c6f 100644 --- a/clients/python/src/taskbroker_client/registry.py +++ b/clients/python/src/taskbroker_client/registry.py @@ -11,6 +11,7 @@ from sentry_protos.taskbroker.v1.taskbroker_pb2 import TaskActivation from sentry_sdk.consts import OP, SPANDATA from sentry_sdk.traces import StreamedSpan +from sentry_sdk.tracing import Span from taskbroker_client.constants import ( DEFAULT_PROCESSING_DEADLINE, @@ -182,7 +183,7 @@ def send_task( span.set_attribute(SPANDATA.MESSAGING_DESTINATION_NAME, activation.namespace) span.set_attribute(SPANDATA.MESSAGING_MESSAGE_ID, activation.id) span.set_attribute(SPANDATA.MESSAGING_SYSTEM, "taskworker") - else: + elif isinstance(span, Span): span.set_data(SPANDATA.MESSAGING_DESTINATION_NAME, activation.namespace) span.set_data(SPANDATA.MESSAGING_MESSAGE_ID, activation.id) span.set_data(SPANDATA.MESSAGING_SYSTEM, "taskworker") diff --git a/clients/python/src/taskbroker_client/sdk.py b/clients/python/src/taskbroker_client/sdk.py index e9071923..833b27b3 100644 --- a/clients/python/src/taskbroker_client/sdk.py +++ b/clients/python/src/taskbroker_client/sdk.py @@ -9,7 +9,7 @@ def start_transaction( - name: str, origin: str, headers: dict[str, Any], sampling_context: dict[str, Any] + name: str, op: str, origin: str, 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 @@ -22,18 +22,21 @@ def start_transaction( return sentry_sdk.traces.start_span( name=name, attributes={ + "sentry.op": op, "sentry.origin": origin, }, ) transaction = sentry_sdk.continue_trace( environ_or_headers=headers, - op="queue.task.taskworker", + op=op, name=name, origin=origin, ) - span = sentry_sdk.start_transaction(transaction, custom_sampling_context=sampling_context) + span = sentry_sdk.start_transaction( + transaction, op, custom_sampling_context=sampling_context + ) except Exception: pass diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py index d3ee9bb4..0b93bc3c 100644 --- a/clients/python/src/taskbroker_client/worker/workerchild.py +++ b/clients/python/src/taskbroker_client/worker/workerchild.py @@ -624,6 +624,7 @@ def _execute_activation( sentry_sdk.isolation_scope(), start_transaction( name=activation.taskname, + op="queue.task.taskworker", origin="taskworker", headers=headers, sampling_context={ @@ -659,7 +660,7 @@ def _execute_activation( SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT, activation.retry_state.attempts ) child_span.set_attribute(SPANDATA.MESSAGING_SYSTEM, "taskworker") - elif isinstance(parent_span, Span): + elif isinstance(child_span, Span): child_span.set_data(SPANDATA.MESSAGING_DESTINATION_NAME, activation.namespace) child_span.set_data(SPANDATA.MESSAGING_MESSAGE_ID, activation.id) child_span.set_data(SPANDATA.MESSAGING_MESSAGE_RECEIVE_LATENCY, latency) From 8272c9fcc4e78a3688c936fec0fd015c2ff5f0c8 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Wed, 22 Jul 2026 17:31:20 +0200 Subject: [PATCH 25/30] remove stray argument --- clients/python/src/taskbroker_client/sdk.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/clients/python/src/taskbroker_client/sdk.py b/clients/python/src/taskbroker_client/sdk.py index 833b27b3..905f8ed4 100644 --- a/clients/python/src/taskbroker_client/sdk.py +++ b/clients/python/src/taskbroker_client/sdk.py @@ -34,9 +34,7 @@ def start_transaction( origin=origin, ) - span = sentry_sdk.start_transaction( - transaction, op, custom_sampling_context=sampling_context - ) + span = sentry_sdk.start_transaction(transaction, custom_sampling_context=sampling_context) except Exception: pass From cd8c9d442fbf274f7a4200903b11a074858cea8b Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Wed, 22 Jul 2026 17:42:34 +0200 Subject: [PATCH 26/30] early return in start_span --- clients/python/src/taskbroker_client/sdk.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/clients/python/src/taskbroker_client/sdk.py b/clients/python/src/taskbroker_client/sdk.py index 905f8ed4..42116e90 100644 --- a/clients/python/src/taskbroker_client/sdk.py +++ b/clients/python/src/taskbroker_client/sdk.py @@ -45,7 +45,6 @@ def start_transaction( def start_span(name: str, op: str, origin: str) -> Span | StreamedSpan | ContextManager[Any]: """Start a span in the currently active trace lifecycle.""" - span = None try: is_span_streaming = has_span_streaming_enabled(sentry_sdk.get_client().options) if is_span_streaming: @@ -57,7 +56,7 @@ def start_span(name: str, op: str, origin: str) -> Span | StreamedSpan | Context }, ) - span = sentry_sdk.start_span( + return sentry_sdk.start_span( op=op, name=name, origin=origin, @@ -65,6 +64,4 @@ def start_span(name: str, op: str, origin: str) -> Span | StreamedSpan | Context except Exception: pass - if span is None: - return nullcontext() - return span + return nullcontext() From fd46b1914dfc44711cc6b6c9386313946ad07688 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Thu, 23 Jul 2026 14:23:20 +0200 Subject: [PATCH 27/30] move attributes into shim --- .../python/src/taskbroker_client/registry.py | 18 +++---- clients/python/src/taskbroker_client/sdk.py | 21 ++++++-- .../taskbroker_client/worker/workerchild.py | 50 +++++++------------ 3 files changed, 41 insertions(+), 48 deletions(-) diff --git a/clients/python/src/taskbroker_client/registry.py b/clients/python/src/taskbroker_client/registry.py index feff2c6f..48e81536 100644 --- a/clients/python/src/taskbroker_client/registry.py +++ b/clients/python/src/taskbroker_client/registry.py @@ -10,8 +10,6 @@ from arroyo.types import BrokerValue, Topic from sentry_protos.taskbroker.v1.taskbroker_pb2 import TaskActivation from sentry_sdk.consts import OP, SPANDATA -from sentry_sdk.traces import StreamedSpan -from sentry_sdk.tracing import Span from taskbroker_client.constants import ( DEFAULT_PROCESSING_DEADLINE, @@ -178,16 +176,12 @@ def send_task( name=activation.taskname, op=OP.QUEUE_PUBLISH, origin="taskworker", - ) as span: - if isinstance(span, StreamedSpan): - span.set_attribute(SPANDATA.MESSAGING_DESTINATION_NAME, activation.namespace) - span.set_attribute(SPANDATA.MESSAGING_MESSAGE_ID, activation.id) - span.set_attribute(SPANDATA.MESSAGING_SYSTEM, "taskworker") - elif isinstance(span, Span): - span.set_data(SPANDATA.MESSAGING_DESTINATION_NAME, activation.namespace) - span.set_data(SPANDATA.MESSAGING_MESSAGE_ID, activation.id) - span.set_data(SPANDATA.MESSAGING_SYSTEM, "taskworker") - + attributes={ + SPANDATA.MESSAGING_DESTINATION_NAME: activation.namespace, + SPANDATA.MESSAGING_MESSAGE_ID: activation.id, + SPANDATA.MESSAGING_SYSTEM: "taskworker", + }, + ): produce_future = self._producer(topic).produce( Topic(name=topic), KafkaPayload(key=None, value=activation.SerializeToString(), headers=[]), diff --git a/clients/python/src/taskbroker_client/sdk.py b/clients/python/src/taskbroker_client/sdk.py index 42116e90..609360e2 100644 --- a/clients/python/src/taskbroker_client/sdk.py +++ b/clients/python/src/taskbroker_client/sdk.py @@ -9,7 +9,12 @@ def start_transaction( - name: str, op: str, origin: str, headers: dict[str, Any], sampling_context: dict[str, Any] + 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 @@ -24,6 +29,7 @@ def start_transaction( attributes={ "sentry.op": op, "sentry.origin": origin, + **attributes, }, ) @@ -35,6 +41,8 @@ def start_transaction( ) 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 @@ -43,7 +51,9 @@ def start_transaction( return span -def start_span(name: str, op: str, origin: str) -> Span | StreamedSpan | ContextManager[Any]: +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.""" try: is_span_streaming = has_span_streaming_enabled(sentry_sdk.get_client().options) @@ -53,14 +63,19 @@ def start_span(name: str, op: str, origin: str) -> Span | StreamedSpan | Context attributes={ "sentry.op": op, "sentry.origin": origin, + **attributes, }, ) - return sentry_sdk.start_span( + span = sentry_sdk.start_span( op=op, name=name, origin=origin, ) + for key, value in attributes.items(): + span.set_data(key, value) + + return span except Exception: pass diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py index 0b93bc3c..9137be05 100644 --- a/clients/python/src/taskbroker_client/worker/workerchild.py +++ b/clients/python/src/taskbroker_client/worker/workerchild.py @@ -32,8 +32,6 @@ ) from sentry_sdk.consts import OP, SPANDATA from sentry_sdk.crons import MonitorStatus, capture_checkin -from sentry_sdk.traces import StreamedSpan -from sentry_sdk.tracing import Span from taskbroker_client.app import import_app from taskbroker_client.constants import CompressionType @@ -626,49 +624,35 @@ def _execute_activation( name=activation.taskname, op="queue.task.taskworker", origin="taskworker", + attributes={ + "taskworker-task.args": args, + "taskworker-task.kwargs": kwargs, + "taskworker-task.id": activation.id, + }, headers=headers, sampling_context={ "taskworker": { "task": activation.taskname, } }, - ) as parent_span, + ), ): - if isinstance(parent_span, StreamedSpan): - parent_span.set_attribute("taskworker-task.args", args) - parent_span.set_attribute("taskworker-task.kwargs", kwargs) - parent_span.set_attribute("taskworker-task.id", activation.id) - elif isinstance(parent_span, Span): - parent_span.set_data( - "taskworker-task", {"args": args, "kwargs": kwargs, "id": activation.id} - ) - task_added_time = activation.received_at.ToDatetime().timestamp() # latency attribute needs to be in milliseconds latency = (time.time() - task_added_time) * 1000 with start_span( - name=activation.taskname, op=OP.QUEUE_PROCESS, origin="taskworker" - ) as child_span: - if isinstance(child_span, StreamedSpan): - child_span.set_attribute( - SPANDATA.MESSAGING_DESTINATION_NAME, activation.namespace - ) - child_span.set_attribute(SPANDATA.MESSAGING_MESSAGE_ID, activation.id) - child_span.set_attribute(SPANDATA.MESSAGING_MESSAGE_RECEIVE_LATENCY, latency) - child_span.set_attribute( - SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT, activation.retry_state.attempts - ) - child_span.set_attribute(SPANDATA.MESSAGING_SYSTEM, "taskworker") - elif isinstance(child_span, Span): - child_span.set_data(SPANDATA.MESSAGING_DESTINATION_NAME, activation.namespace) - child_span.set_data(SPANDATA.MESSAGING_MESSAGE_ID, activation.id) - child_span.set_data(SPANDATA.MESSAGING_MESSAGE_RECEIVE_LATENCY, latency) - child_span.set_data( - SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT, activation.retry_state.attempts - ) - child_span.set_data(SPANDATA.MESSAGING_SYSTEM, "taskworker") - + name=activation.taskname, + op=OP.QUEUE_PROCESS, + origin="taskworker", + attributes={ + SPANDATA.MESSAGING_DESTINATION_NAME: activation.namespace, + SPANDATA.MESSAGING_MESSAGE_ID: activation.id, + SPANDATA.MESSAGING_MESSAGE_RECEIVE_LATENCY: latency, + SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT: activation.retry_state.attempts, + SPANDATA.MESSAGING_SYSTEM: "taskworker", + }, + ): # TODO(taskworker) remove this when doing cleanup # The `__start_time` parameter is spliced into task parameters by # sentry.celery.SentryTask._add_metadata and needs to be removed From 106634c39f53472f93ea50e02b9e118cf8859764 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Thu, 23 Jul 2026 15:01:56 +0200 Subject: [PATCH 28/30] bump sentry-sdk to when span streaming is fully available --- clients/python/pyproject.toml | 2 +- uv.lock | 8 ++++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/clients/python/pyproject.toml b/clients/python/pyproject.toml index f15b552d..4e9168fb 100644 --- a/clients/python/pyproject.toml +++ b/clients/python/pyproject.toml @@ -5,7 +5,7 @@ description = "Taskbroker python client and worker runtime" readme = "README.md" dependencies = [ "sentry-arroyo>=2.41.0", - "sentry-sdk[http2]>=2.52.0", + "sentry-sdk[http2]==2.66.1", "sentry-protos>=0.26.1", "confluent_kafka>=2.3.0", "cronsim>=2.6", diff --git a/uv.lock b/uv.lock index b0e85a2f..740bdf0b 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 3 +revision = 2 requires-python = ">=3.11" resolution-markers = [ "sys_platform == 'darwin' or sys_platform == 'linux'", @@ -703,14 +703,14 @@ wheels = [ [[package]] name = "sentry-sdk" -version = "2.66.0" +version = "2.66.1" source = { registry = "https://pypi.devinfra.sentry.io/simple" } dependencies = [ { name = "certifi", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" }, { name = "urllib3", marker = "sys_platform == 'darwin' or sys_platform == 'linux'" }, ] wheels = [ - { url = "https://pypi.devinfra.sentry.io/wheels/sentry_sdk-2.66.0-py3-none-any.whl", hash = "sha256:096136c214c602be2b323524d30755dc5b30ec5a218a206207f33b12c05c6f11" }, + { url = "https://pypi.devinfra.sentry.io/wheels/sentry_sdk-2.66.1-py3-none-any.whl", hash = "sha256:86002793161d9a95ef04bdd8d442e9bfece5d989b755f05d6360215094a7aff6" }, ] [package.optional-dependencies] @@ -856,7 +856,7 @@ requires-dist = [ { name = "redis-py-cluster", marker = "extra == 'cluster'", specifier = ">=2.1.0" }, { name = "sentry-arroyo", specifier = ">=2.41.0" }, { name = "sentry-protos", specifier = ">=0.26.1" }, - { name = "sentry-sdk", extras = ["http2"], specifier = ">=2.52.0" }, + { name = "sentry-sdk", extras = ["http2"], specifier = "==2.66.1" }, { name = "setuptools", marker = "extra == 'examples'", specifier = ">=80.0" }, { name = "zstandard", specifier = ">=0.18.0" }, ] From 6b478e6ea5f6379de03413ab98a2dec66b0b7540 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Thu, 23 Jul 2026 15:02:44 +0200 Subject: [PATCH 29/30] use greater than or equal --- clients/python/pyproject.toml | 2 +- uv.lock | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/clients/python/pyproject.toml b/clients/python/pyproject.toml index 4e9168fb..3493889a 100644 --- a/clients/python/pyproject.toml +++ b/clients/python/pyproject.toml @@ -5,7 +5,7 @@ description = "Taskbroker python client and worker runtime" readme = "README.md" dependencies = [ "sentry-arroyo>=2.41.0", - "sentry-sdk[http2]==2.66.1", + "sentry-sdk[http2]>=2.66.1", "sentry-protos>=0.26.1", "confluent_kafka>=2.3.0", "cronsim>=2.6", diff --git a/uv.lock b/uv.lock index 740bdf0b..22886015 100644 --- a/uv.lock +++ b/uv.lock @@ -856,7 +856,7 @@ requires-dist = [ { name = "redis-py-cluster", marker = "extra == 'cluster'", specifier = ">=2.1.0" }, { name = "sentry-arroyo", specifier = ">=2.41.0" }, { name = "sentry-protos", specifier = ">=0.26.1" }, - { name = "sentry-sdk", extras = ["http2"], specifier = "==2.66.1" }, + { name = "sentry-sdk", extras = ["http2"], specifier = ">=2.66.1" }, { name = "setuptools", marker = "extra == 'examples'", specifier = ">=80.0" }, { name = "zstandard", specifier = ">=0.18.0" }, ] From 7770df5d5021f6717d4164b1f39002f5bd7597c5 Mon Sep 17 00:00:00 2001 From: Alexander Alderman Webb Date: Thu, 23 Jul 2026 15:56:26 +0200 Subject: [PATCH 30/30] do not add args and kwargs attributes to StreamedSpan --- .../python/src/taskbroker_client/worker/workerchild.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py index 9137be05..be9736e8 100644 --- a/clients/python/src/taskbroker_client/worker/workerchild.py +++ b/clients/python/src/taskbroker_client/worker/workerchild.py @@ -32,6 +32,7 @@ ) 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 @@ -625,8 +626,6 @@ def _execute_activation( op="queue.task.taskworker", origin="taskworker", attributes={ - "taskworker-task.args": args, - "taskworker-task.kwargs": kwargs, "taskworker-task.id": activation.id, }, headers=headers, @@ -635,8 +634,13 @@ 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) + task_added_time = activation.received_at.ToDatetime().timestamp() # latency attribute needs to be in milliseconds latency = (time.time() - task_added_time) * 1000