diff --git a/docs/explanations/events.md b/docs/explanations/events.md index 7823408f1..3636617e9 100644 --- a/docs/explanations/events.md +++ b/docs/explanations/events.md @@ -34,18 +34,9 @@ every time the engine: In the latter case, information about the error is also included. -## Correlation ID +## Trace Context - -When controlling plans programmatically, it can be useful to verify that event model documents really are related to -the plan you just asked the worker to run. The worker will therefore bundle a correlation ID into the headers of -messages containing documents. - -see also [`Microsoft Playbook on Correlation IDs`](https://microsoft.github.io/code-with-engineering-playbook/observability/correlation-id/) - -ActiveMQ will give this header a different name depending on the protocol you use. - -| Protocol | Header name | -|----------|------------------| -| JMS | jms_correlationID| -| STOMP | correlation-id | +When controlling plans programmatically, it can be useful to verify that events really are related to +the plan you just asked the worker to run. Every event carries the task ID in its body, and the worker also +propagates the [OpenTelemetry](https://opentelemetry.io/) trace context in the message headers +(the W3C `traceparent` header), so events can be tied to the request that started the task. diff --git a/docs/reference/asyncapi.yaml b/docs/reference/asyncapi.yaml index 097a7e039..6416b5e20 100644 --- a/docs/reference/asyncapi.yaml +++ b/docs/reference/asyncapi.yaml @@ -31,7 +31,7 @@ channels: - $ref: "#/components/messages/taggedResourceStream" - $ref: "#/components/messages/taggedDatumStream" components: - messages: # Defined as Components to allow referencing from other APIs. # TODO: Define protocol specific correlationId bindings + messages: # Defined as Components to allow referencing from other APIs. taggedStartDocument: messageId: runStart summary: Indicates the start of a Bluesky run based on a previously specified Plan; describes the initial position/metadata of a scan and its involved devices @@ -183,9 +183,6 @@ components: replyDestination: description: Name of the temporary queue, specified by the caller where they will listen for response messages to requests submitted on the worker.X channels type: string - correlationId: - description: Unique identifier of an exchange supplied by its originator - type: string workerState: description: The state of the Worker. enum: diff --git a/docs/reference/openapi.yaml b/docs/reference/openapi.yaml index c17734bbe..e72eda45f 100644 --- a/docs/reference/openapi.yaml +++ b/docs/reference/openapi.yaml @@ -388,9 +388,6 @@ components: - $ref: '#/components/schemas/TaskError' - type: 'null' title: Outcome - request_id: - title: Request Id - type: string task: $ref: '#/components/schemas/Task' task_id: @@ -459,7 +456,7 @@ info: name: Apache 2.0 url: https://www.apache.org/licenses/LICENSE-2.0.html title: BlueAPI Control - version: 1.6.0 + version: 1.6.1 openapi: 3.1.0 paths: /api/v1/devices: diff --git a/src/blueapi/client/client.py b/src/blueapi/client/client.py index 2105ac64e..3f5d80109 100644 --- a/src/blueapi/client/client.py +++ b/src/blueapi/client/client.py @@ -407,11 +407,7 @@ def inner_on_event(event: AnyEvent, ctx: MessageContext) -> None: log.error( f"Callback ({cb}) failed for event: {event}", exc_info=e ) - if ( - isinstance(event, WorkerEvent) - and (event.is_complete()) - and (ctx.correlation_id == task_id) - ): + if isinstance(event, WorkerEvent) and event.is_complete(): if event.task_status is None: complete.set_exception( BlueskyRemoteControlError( diff --git a/src/blueapi/config.py b/src/blueapi/config.py index f65dd65f0..524ea8aba 100644 --- a/src/blueapi/config.py +++ b/src/blueapi/config.py @@ -335,7 +335,7 @@ class ApplicationConfig(BlueapiBaseModel): """ #: API version to publish in OpenAPI schema - REST_API_VERSION: ClassVar[str] = "1.6.0" + REST_API_VERSION: ClassVar[str] = "1.6.1" LICENSE_INFO: ClassVar[dict[str, str]] = { "name": "Apache 2.0", diff --git a/src/blueapi/core/event.py b/src/blueapi/core/event.py index 9308f8ef4..b3781d9df 100644 --- a/src/blueapi/core/event.py +++ b/src/blueapi/core/event.py @@ -19,12 +19,12 @@ class EventStream(ABC, Generic[E, S]): """ @abstractmethod - def subscribe(self, __callback: Callable[[E, str | None], None]) -> S: + def subscribe(self, __callback: Callable[[E], None]) -> S: """ Subscribe to new events with a callback Args: - __callback: What to do with each event, optionally takes a correlation id + __callback: What to do with each event Returns: S: A unique token representing the subscription @@ -51,14 +51,14 @@ class EventPublisher(EventStream[E, int]): Simple Observable that can be fed values to publish """ - _subscriptions: dict[int, Callable[[E, str | None], None]] + _subscriptions: dict[int, Callable[[E], None]] _count: itertools.count def __init__(self) -> None: self._subscriptions = {} self._count = itertools.count() - def subscribe(self, callback: Callable[[E, str | None], None]) -> int: + def subscribe(self, callback: Callable[[E], None]) -> int: sub_id = next(self._count) self._subscriptions[sub_id] = callback return sub_id @@ -69,19 +69,17 @@ def unsubscribe(self, subscription: int) -> None: def unsubscribe_all(self) -> None: self._subscriptions = {} - def publish(self, event: E, correlation_id: str | None = None) -> None: + def publish(self, event: E) -> None: """ Publish a new event to all subscribers Args: event: The event to publish - correlation_id: An optional ID that may be used to correlate this - event with other events """ errs = [] for callback in list(self._subscriptions.values()): try: - callback(event, correlation_id) + callback(event) except Exception as e: errs.append(e) if errs: diff --git a/src/blueapi/service/interface.py b/src/blueapi/service/interface.py index 3c67a8dd4..be6267886 100644 --- a/src/blueapi/service/interface.py +++ b/src/blueapi/service/interface.py @@ -189,11 +189,9 @@ def _publish_event_streams( def _publish_event_stream(stream: EventStream, destination: DestinationBase) -> None: - def forward_message(event: Any, correlation_id: str | None) -> None: + def forward_message(event: Any) -> None: if (stomp_client_ref := stomp_client()) is not None: - stomp_client_ref.send( - destination, event, None, correlation_id=correlation_id - ) + stomp_client_ref.send(destination, event, None) stream.subscribe(forward_message) @@ -288,9 +286,7 @@ def begin_task( ) subscribers.append((active_context.run_engine, tiled_writer_token)) - def remove_callback_when_task_finished( - event: WorkerEvent, correlation_id: str | None - ) -> None: + def remove_callback_when_task_finished(event: WorkerEvent) -> None: if ( event.task_status and event.task_status.task_id == task.task_id @@ -373,11 +369,7 @@ class SubHandles: def pipe_events(sender: Connection) -> SubHandles: tw = worker() - def handler( - worker_event: WorkerEvent | DataEvent | ProgressEvent, - _cor_id: str | None, - ) -> None: - + def handler(worker_event: WorkerEvent | DataEvent | ProgressEvent) -> None: try: sender.send(worker_event) except BrokenPipeError: diff --git a/src/blueapi/worker/task_worker.py b/src/blueapi/worker/task_worker.py index 219a1c1ca..c4a291437 100644 --- a/src/blueapi/worker/task_worker.py +++ b/src/blueapi/worker/task_worker.py @@ -16,11 +16,9 @@ setup_tracing, start_as_current_span, ) -from opentelemetry.baggage import get_baggage from opentelemetry.context import Context, get_current from opentelemetry.trace import SpanKind from pydantic import Field -from pydantic.json_schema import SkipJsonSchema from blueapi.core import ( OTLP_EXPORT_ENABLED, @@ -67,7 +65,6 @@ class TrackableTask(BlueapiBaseModel): task_id: str task: Task - request_id: str | SkipJsonSchema[None] = None is_complete: bool = False is_pending: bool = True errors: list[str] = Field(default_factory=list) @@ -275,16 +272,7 @@ def submit_task(self, task: Task) -> str: task_id: str = str(uuid.uuid4()) task.metadata["blueapi_task_id"] = task_id add_span_attributes({"TaskId": task_id}) - request_id = get_baggage("correlation_id") - # If request id is not a string, we do not pass it into a TrackableTask - if not isinstance(request_id, str): - LOGGER.warning(f"Invalid correlation id detected: {request_id}") - request_id = None - trackable_task = TrackableTask( - task_id=task_id, - request_id=request_id, - task=task, - ) + trackable_task = TrackableTask(task_id=task_id, task=task) self._pending_tasks[task_id] = trackable_task return task_id @@ -303,7 +291,7 @@ def _submit_trackable_task(self, trackable_task: TrackableTask) -> None: task_started = Event() - def mark_task_as_started(event: WorkerEvent, _: str | None) -> None: + def mark_task_as_started(event: WorkerEvent) -> None: if ( event.task_status is not None and event.task_status.task_id == trackable_task.task_id @@ -544,7 +532,6 @@ def _report_status( task_failed=bool(self._current.errors), result=self._current.outcome, ) - correlation_id = self._current.task_id add_span_attributes( { "task_id": self._current.task_id, @@ -554,7 +541,6 @@ def _report_status( ) else: task_status = None - correlation_id = None event = WorkerEvent( state=self._state, @@ -562,7 +548,7 @@ def _report_status( errors=errors, warnings=warnings, ) - self._worker_events.publish(event, correlation_id) + self._worker_events.publish(event) def _on_document(self, name: str, document: Mapping[str, Any]) -> None: if self._current is not None: @@ -585,12 +571,10 @@ def _on_document(self, name: str, document: Mapping[str, Any]) -> None: } ) - correlation_id = self._current.request_id self._data_events.publish( DataEvent( name=name, task_id=self._current.task_id, doc=document - ), - correlation_id, + ) ) else: raise ValueError( @@ -670,8 +654,7 @@ def _publish_status_snapshot(self) -> None: ProgressEvent( task_id=self._current.task_id, statuses=self._status_snapshot, - ), - self._current.task_id, + ) ) diff --git a/tests/unit_tests/cli/test_cli.py b/tests/unit_tests/cli/test_cli.py index f858e6ac0..3bf3b5050 100644 --- a/tests/unit_tests/cli/test_cli.py +++ b/tests/unit_tests/cli/test_cli.py @@ -291,7 +291,6 @@ def mock_events(topic: MessageTopic, callback: Callable[[Any, Any], Any]): if topic.name != "public.worker.event": return ctx = Mock() - ctx.correlation_id = task_id callback( WorkerEvent( state=WorkerState.RUNNING, diff --git a/tests/unit_tests/client/test_client.py b/tests/unit_tests/client/test_client.py index e7e31f930..eef5eade9 100644 --- a/tests/unit_tests/client/test_client.py +++ b/tests/unit_tests/client/test_client.py @@ -48,7 +48,6 @@ def mock_events() -> EventBusClient: mock_events = MagicMock(spec=EventBusClient) ctx = Mock() - ctx.correlation_id = "foo" mock_events.subscribe_to_all_events = lambda on_event: on_event(ctx, COMPLETE_EVENT) return mock_events @@ -374,7 +373,6 @@ def test_run_stomp_sets_up_control( mock_rest.create_task.return_value = TaskResponse(task_id="foo") mock_rest.update_worker_task.return_value = TaskResponse(task_id="foo") ctx = Mock() - ctx.correlation_id = "foo" mock_events.subscribe_to_all_events = lambda on_event: on_event(COMPLETE_EVENT, ctx) client_with_events.run_stomp( @@ -395,7 +393,6 @@ def test_run_stomp_fails_on_failing_event( mock_rest.update_worker_task.return_value = TaskResponse(task_id="foo") ctx = Mock() - ctx.correlation_id = "foo" mock_events.subscribe_to_all_events = lambda on_event: on_event(FAILED_EVENT, ctx) on_event = Mock() @@ -438,7 +435,6 @@ def test_run_stomp_calls_event_callback( mock_rest.update_worker_task.return_value = TaskResponse(task_id="foo") ctx = Mock() - ctx.correlation_id = "foo" def callback(on_event: Callable[[AnyEvent, MessageContext], None]): on_event(test_event, ctx) @@ -480,7 +476,6 @@ def test_run_stomp_ignores_non_matching_events( mock_rest.update_worker_task.return_value = TaskResponse(task_id="foo") # type: ignore ctx = Mock() - ctx.correlation_id = "foo" def callback(on_event: Callable[[AnyEvent, MessageContext], None]): on_event(test_event, ctx) # type: ignore @@ -715,7 +710,6 @@ def test_client_callbacks( mock_rest.update_worker_task.return_value = TaskResponse(task_id="foo") ctx = Mock() - ctx.correlation_id = "foo" def subscribe(on_event: Callable[[AnyEvent, MessageContext], None]): on_event(test_event, ctx) @@ -743,7 +737,6 @@ def test_client_callback_failures( mock_rest.update_worker_task.return_value = TaskResponse(task_id="foo") ctx = Mock() - ctx.correlation_id = "foo" evt = DataEvent(name="start", doc={}, task_id="foo") diff --git a/tests/unit_tests/core/test_event.py b/tests/unit_tests/core/test_event.py index 2570d9ec7..a6515060b 100644 --- a/tests/unit_tests/core/test_event.py +++ b/tests/unit_tests/core/test_event.py @@ -24,7 +24,7 @@ def publisher() -> EventPublisher[MyEvent]: def test_publishes_event(publisher: EventPublisher[MyEvent]) -> None: event = MyEvent("a") f: Future = Future() - publisher.subscribe(lambda r, _: f.set_result(r)) + publisher.subscribe(lambda r: f.set_result(r)) publisher.publish(event) assert f.result(timeout=_TIMEOUT) == event @@ -33,8 +33,8 @@ def test_multi_subscriber(publisher: EventPublisher[MyEvent]) -> None: event = MyEvent("a") f1: Future = Future() f2: Future = Future() - publisher.subscribe(lambda r, _: f1.set_result(r)) - publisher.subscribe(lambda r, _: f2.set_result(r)) + publisher.subscribe(lambda r: f1.set_result(r)) + publisher.subscribe(lambda r: f2.set_result(r)) publisher.publish(event) assert f1.result(timeout=_TIMEOUT) == f2.result(timeout=_TIMEOUT) == event @@ -44,11 +44,11 @@ def test_can_unsubscribe(publisher: EventPublisher[MyEvent]) -> None: event_b = MyEvent("b") event_c = MyEvent("c") q: Queue = Queue() - sub = publisher.subscribe(lambda r, _: q.put(r)) + sub = publisher.subscribe(lambda r: q.put(r)) publisher.publish(event_a) publisher.unsubscribe(sub) publisher.publish(event_b) - publisher.subscribe(lambda r, _: q.put(r)) + publisher.subscribe(lambda r: q.put(r)) publisher.publish(event_c) assert list(_drain(q)) == [event_a, event_c] @@ -58,28 +58,18 @@ def test_can_unsubscribe_all(publisher: EventPublisher[MyEvent]) -> None: event_b = MyEvent("b") event_c = MyEvent("c") q: Queue = Queue() - publisher.subscribe(lambda r, _: q.put(r)) - publisher.subscribe(lambda r, _: q.put(r)) + publisher.subscribe(lambda r: q.put(r)) + publisher.subscribe(lambda r: q.put(r)) publisher.publish(event_a) publisher.unsubscribe_all() publisher.publish(event_b) - publisher.subscribe(lambda r, _: q.put(r)) + publisher.subscribe(lambda r: q.put(r)) publisher.publish(event_c) assert list(_drain(q)) == [event_a, event_a, event_c] -def test_correlation_id(publisher: EventPublisher[MyEvent]) -> None: - event = MyEvent("a") - correlation_id = "foobar" - f: Future = Future() - publisher.subscribe(lambda _, c: f.set_result(c)) - publisher.publish(event, correlation_id) - assert f.result(timeout=_TIMEOUT) == correlation_id - - def test_callback_exceptions_are_contained(publisher: EventPublisher[MyEvent]): event = MyEvent("foo") - c_id = "bar" # First call should raise exception, next should be fine handler = mock.Mock(side_effect=[ValueError("Bad Event"), ()]) @@ -87,10 +77,10 @@ def test_callback_exceptions_are_contained(publisher: EventPublisher[MyEvent]): publisher.subscribe(handler) with pytest.RaisesGroup(ValueError): - publisher.publish(event, c_id) + publisher.publish(event) # Both handlers should be called but the exception should still be raised - handler.assert_has_calls([mock.call(event, c_id), mock.call(event, c_id)]) + handler.assert_has_calls([mock.call(event), mock.call(event)]) def _drain(queue: Queue) -> Iterable: diff --git a/tests/unit_tests/service/test_interface.py b/tests/unit_tests/service/test_interface.py index e56e98c2e..f00adff69 100644 --- a/tests/unit_tests/service/test_interface.py +++ b/tests/unit_tests/service/test_interface.py @@ -4,7 +4,7 @@ from inspect import isawaitable from multiprocessing.connection import Connection as PipeConnection from typing import Any -from unittest.mock import ANY, MagicMock, Mock, patch +from unittest.mock import MagicMock, Mock, patch import pytest from bluesky.protocols import Stoppable @@ -392,7 +392,6 @@ def test_get_task_by_id( assert interface.get_task_by_id(task_id) == TrackableTask.model_construct( task_id=task_id, - request_id=ANY, task=Task( name="my_plan", params={}, @@ -424,7 +423,6 @@ def test_submit_task_inserts_metadata(context_mock: MagicMock): assert interface.get_task_by_id(task_id) == TrackableTask.model_construct( task_id=task_id, - request_id=ANY, task=Task( name="my_plan", params={}, @@ -464,8 +462,7 @@ def test_remove_tiled_subscriber(worker, context, from_uri, writer): task_failed=False, result=None, ), - ), - "c_id", + ) ) context().run_engine.unsubscribe.assert_not_called() worker().worker_events.unsubscribe.assert_not_called() @@ -479,8 +476,7 @@ def test_remove_tiled_subscriber(worker, context, from_uri, writer): task_failed=False, result=TaskResult(result=None, type="NoneType"), ), - ), - "c_id", + ) ) context().run_engine.unsubscribe.assert_called_once_with(17) worker().worker_events.unsubscribe.assert_called_once_with(42) @@ -867,7 +863,7 @@ def test_pipe_events(mock_worker: Mock): handler = worker.worker_events.subscribe.call_args[0][0] evt = Mock() - handler(evt, "ignored correlation id") + handler(evt) tx.send.assert_called_once_with(evt) @@ -883,7 +879,7 @@ def test_pipe_events_ignores_broken_pipe(mock_worker: Mock): tx.send.side_effect = BrokenPipeError() # ensure that exceptions are not raised - handler(Mock(), "ignored correlation id") + handler(Mock()) @patch("blueapi.service.interface.worker") diff --git a/tests/unit_tests/service/test_rest_api.py b/tests/unit_tests/service/test_rest_api.py index 4315eecf6..e15b926cc 100644 --- a/tests/unit_tests/service/test_rest_api.py +++ b/tests/unit_tests/service/test_rest_api.py @@ -440,7 +440,6 @@ def test_get_tasks(mock_runner: Mock, client: TestClient) -> None: "errors": [], "is_complete": False, "is_pending": True, - "request_id": None, "task": { "name": "sleep", "params": {"time": 0.0}, @@ -453,7 +452,6 @@ def test_get_tasks(mock_runner: Mock, client: TestClient) -> None: "errors": [], "is_complete": False, "is_pending": True, - "request_id": None, "task": { "name": "first_task", "params": {}, @@ -485,7 +483,6 @@ def test_get_tasks_by_status(mock_runner: Mock, client: TestClient) -> None: "errors": [], "is_complete": True, "is_pending": False, - "request_id": None, "task": { "name": "third_task", "params": {}, @@ -649,7 +646,6 @@ def test_get_task(mock_runner: Mock, client: TestClient): "errors": [], "is_complete": False, "is_pending": True, - "request_id": None, "task": { "name": "third_task", "params": {}, @@ -704,7 +700,6 @@ def test_get_all_tasks(mock_runner: Mock, client: TestClient): }, "is_complete": False, "is_pending": True, - "request_id": None, "outcome": None, "errors": [], } diff --git a/tests/unit_tests/worker/test_task_worker.py b/tests/unit_tests/worker/test_task_worker.py index 044668d93..65470bef0 100644 --- a/tests/unit_tests/worker/test_task_worker.py +++ b/tests/unit_tests/worker/test_task_worker.py @@ -5,7 +5,7 @@ from concurrent.futures import Future from queue import Full from typing import Any, TypeVar -from unittest.mock import ANY, MagicMock, Mock, patch +from unittest.mock import MagicMock, Mock, patch import pydantic import pytest @@ -183,9 +183,7 @@ def test_submit_task( assert worker.get_tasks() == [] task_id = worker.submit_task(_SIMPLE_TASK) assert worker.get_tasks() == [ - TrackableTask.model_construct( - task_id=task_id, request_id=ANY, task=_SIMPLE_TASK - ) + TrackableTask.model_construct(task_id=task_id, task=_SIMPLE_TASK) ] @@ -193,18 +191,12 @@ def test_submit_multiple_tasks(worker: TaskWorker) -> None: assert worker.get_tasks() == [] task_id_1 = worker.submit_task(_SIMPLE_TASK) assert worker.get_tasks() == [ - TrackableTask.model_construct( - task_id=task_id_1, request_id=ANY, task=_SIMPLE_TASK - ) + TrackableTask.model_construct(task_id=task_id_1, task=_SIMPLE_TASK) ] task_id_2 = worker.submit_task(_LONG_TASK) assert worker.get_tasks() == [ - TrackableTask.model_construct( - task_id=task_id_1, request_id=ANY, task=_SIMPLE_TASK - ), - TrackableTask.model_construct( - task_id=task_id_2, request_id=ANY, task=_LONG_TASK - ), + TrackableTask.model_construct(task_id=task_id_1, task=_SIMPLE_TASK), + TrackableTask.model_construct(task_id=task_id_2, task=_LONG_TASK), ] @@ -217,16 +209,12 @@ def test_stop_with_task_pending(inert_worker: TaskWorker) -> None: def test_restart_leaves_task_pending(worker: TaskWorker) -> None: task_id = worker.submit_task(_SIMPLE_TASK) assert worker.get_tasks() == [ - TrackableTask.model_construct( - task_id=task_id, request_id=ANY, task=_SIMPLE_TASK - ) + TrackableTask.model_construct(task_id=task_id, task=_SIMPLE_TASK) ] worker.stop() worker.start() assert worker.get_tasks() == [ - TrackableTask.model_construct( - task_id=task_id, request_id=ANY, task=_SIMPLE_TASK - ) + TrackableTask.model_construct(task_id=task_id, task=_SIMPLE_TASK) ] @@ -234,24 +222,18 @@ def test_submit_before_start_pending(inert_worker: TaskWorker) -> None: task_id = inert_worker.submit_task(_SIMPLE_TASK) inert_worker.start() assert inert_worker.get_tasks() == [ - TrackableTask.model_construct( - task_id=task_id, request_id=ANY, task=_SIMPLE_TASK - ) + TrackableTask.model_construct(task_id=task_id, task=_SIMPLE_TASK) ] inert_worker.stop() assert inert_worker.get_tasks() == [ - TrackableTask.model_construct( - task_id=task_id, request_id=ANY, task=_SIMPLE_TASK - ) + TrackableTask.model_construct(task_id=task_id, task=_SIMPLE_TASK) ] def test_clear_task(worker: TaskWorker) -> None: task_id = worker.submit_task(_SIMPLE_TASK) assert worker.get_tasks() == [ - TrackableTask.model_construct( - task_id=task_id, request_id=ANY, task=_SIMPLE_TASK - ) + TrackableTask.model_construct(task_id=task_id, task=_SIMPLE_TASK) ] assert worker.clear_task(task_id) assert worker.get_tasks() == [] @@ -556,7 +538,7 @@ def take_events( events: list[E] = [] future: Future[list[E]] = Future() - def on_event(event: E, event_id: str | None) -> None: + def on_event(event: E) -> None: events.append(event) if cutoff_predicate(event): future.set_result(events) @@ -592,7 +574,7 @@ def take_events_from_streams( events: list[Any] = [] future: Future[list[Any]] = Future() - def on_event(event: Any, event_id: str | None) -> None: + def on_event(event: Any) -> None: print(event) events.append(event) if cutoff_predicate(event): @@ -686,19 +668,6 @@ def test_submit_task_span_ok( worker.submit_task(_SIMPLE_TASK) -@patch("blueapi.worker.task_worker.get_baggage") -def test_submit_task_with_correlation_id( - mock_get_baggage: Mock, - exporter: JsonObjectSpanExporter, - worker: TaskWorker, -) -> None: - mock_get_baggage.return_value = "foo" - task_id = worker.submit_task(_SIMPLE_TASK) - trackable_task = worker.get_task_by_id(task_id) - assert trackable_task is not None - assert trackable_task.request_id == "foo" - - def test_clear_task_span_ok( exporter: JsonObjectSpanExporter, worker: TaskWorker ) -> None: