diff --git a/sentry_sdk/integrations/arq.py b/sentry_sdk/integrations/arq.py index d318152f88..c82a01da31 100644 --- a/sentry_sdk/integrations/arq.py +++ b/sentry_sdk/integrations/arq.py @@ -6,7 +6,6 @@ from sentry_sdk.integrations.logging import ignore_logger from sentry_sdk.scope import should_send_default_pii from sentry_sdk.traces import SegmentNameSource -from sentry_sdk.tracing import Transaction, TransactionSource from sentry_sdk.tracing_utils import has_span_streaming_enabled from sentry_sdk.utils import ( SENSITIVE_DATA_SUBSTITUTE, @@ -79,21 +78,15 @@ async def _sentry_enqueue_job( if client.get_integration(ArqIntegration) is None: return await old_enqueue_job(self, function, *args, **kwargs) - if has_span_streaming_enabled(client.options): - if sentry_sdk.traces.get_current_span() is None: - return await old_enqueue_job(self, function, *args, **kwargs) - - with sentry_sdk.traces.start_span( - name=function, - attributes={ - "sentry.op": OP.QUEUE_SUBMIT_ARQ, - "sentry.origin": ArqIntegration.origin, - }, - ): - return await old_enqueue_job(self, function, *args, **kwargs) + if sentry_sdk.traces.get_current_span() is None: + return await old_enqueue_job(self, function, *args, **kwargs) - with sentry_sdk.start_span( - op=OP.QUEUE_SUBMIT_ARQ, name=function, origin=ArqIntegration.origin + with sentry_sdk.traces.start_span( + name=function, + attributes={ + "sentry.op": OP.QUEUE_SUBMIT_ARQ, + "sentry.origin": ArqIntegration.origin, + }, ): return await old_enqueue_job(self, function, *args, **kwargs) @@ -113,34 +106,20 @@ async def _sentry_run_job(self: "Worker", job_id: str, score: int) -> None: scope._name = "arq" scope.clear_breadcrumbs() - if has_span_streaming_enabled(client.options): - with sentry_sdk.traces.start_span( - name="unknown arq task", - attributes={ - "sentry.op": OP.QUEUE_TASK_ARQ, - "sentry.origin": ArqIntegration.origin, - "sentry.segment.name.source": SegmentNameSource.TASK, - SPANDATA.MESSAGING_MESSAGE_ID: job_id, - }, - parent_span=None, - ) as span: - if self.queue_name is not None: - span.set_attribute( - SPANDATA.MESSAGING_DESTINATION_NAME, self.queue_name - ) - return await old_run_job(self, job_id, score) - - transaction = Transaction( + with sentry_sdk.traces.start_span( name="unknown arq task", - status="ok", - op=OP.QUEUE_TASK_ARQ, - source=TransactionSource.TASK, - origin=ArqIntegration.origin, - ) - - with sentry_sdk.start_transaction(transaction) as span: + attributes={ + "sentry.op": OP.QUEUE_TASK_ARQ, + "sentry.origin": ArqIntegration.origin, + "sentry.segment.name.source": SegmentNameSource.TASK, + SPANDATA.MESSAGING_MESSAGE_ID: job_id, + }, + parent_span=None, + ) as span: if self.queue_name is not None: - span.set_data(SPANDATA.MESSAGING_DESTINATION_NAME, self.queue_name) + span.set_attribute( + SPANDATA.MESSAGING_DESTINATION_NAME, self.queue_name + ) return await old_run_job(self, job_id, score) Worker.run_job = _sentry_run_job diff --git a/tests/integrations/arq/test_arq.py b/tests/integrations/arq/test_arq.py index 5601c6dea3..1c1ea43a5c 100644 --- a/tests/integrations/arq/test_arq.py +++ b/tests/integrations/arq/test_arq.py @@ -10,7 +10,7 @@ from fakeredis.aioredis import FakeRedis import sentry_sdk -from sentry_sdk import get_client, start_transaction +from sentry_sdk import get_client from sentry_sdk.consts import SPANDATA from sentry_sdk.integrations.arq import ArqIntegration from tests.integrations.utils import DATA_COLLECTION_QUEUES_CASES @@ -44,7 +44,6 @@ def info(self, section): @pytest.fixture def init_arq(sentry_init): def inner( - span_streaming, cls_functions=None, cls_cron_jobs=None, kw_functions=None, @@ -65,7 +64,7 @@ def inner( "integrations": [ArqIntegration()], "traces_sample_rate": 1.0, "send_default_pii": True, - "trace_lifecycle": "stream" if span_streaming else "static", + "trace_lifecycle": "stream", } sentry_init_kwargs.update(init_kwargs or {}) sentry_init(**sentry_init_kwargs) @@ -94,7 +93,6 @@ class WorkerSettings: @pytest.fixture def init_arq_with_dict_settings(sentry_init): def inner( - span_streaming, cls_functions=None, cls_cron_jobs=None, kw_functions=None, @@ -115,7 +113,7 @@ def inner( "integrations": [ArqIntegration()], "traces_sample_rate": 1.0, "send_default_pii": True, - "trace_lifecycle": "stream" if span_streaming else "static", + "trace_lifecycle": "stream", } sentry_init_kwargs.update(init_kwargs or {}) sentry_init(**sentry_init_kwargs) @@ -147,7 +145,6 @@ def init_arq_with_kwarg_settings(sentry_init): """Test fixture that passes settings_cls as keyword argument only.""" def inner( - span_streaming, cls_functions=None, cls_cron_jobs=None, kw_functions=None, @@ -168,7 +165,7 @@ def inner( "integrations": [ArqIntegration()], "traces_sample_rate": 1.0, "send_default_pii": True, - "trace_lifecycle": "stream" if span_streaming else "static", + "trace_lifecycle": "stream", } sentry_init_kwargs.update(init_kwargs or {}) sentry_init(**sentry_init_kwargs) @@ -200,8 +197,10 @@ class WorkerSettings: "init_arq_settings", ["init_arq", "init_arq_with_dict_settings", "init_arq_with_kwarg_settings"], ) -@pytest.mark.parametrize("span_streaming", [True, False]) -async def test_job_result(init_arq_settings, request, span_streaming): +async def test_job_result( + init_arq_settings, + request, +): async def increase(ctx, num): return num + 1 @@ -209,7 +208,7 @@ async def increase(ctx, num): increase.__qualname__ = increase.__name__ - pool, worker = init_fixture_method(span_streaming, [increase]) + pool, worker = init_fixture_method([increase]) job = await pool.enqueue_job("increase", 3) @@ -227,13 +226,10 @@ async def increase(ctx, num): @pytest.mark.parametrize( "init_arq_settings", ["init_arq", "init_arq_with_dict_settings"] ) -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_job_retry( - capture_events, capture_items, init_arq_settings, request, - span_streaming, ): async def retry_job(ctx): if ctx["job_try"] < 2: @@ -243,52 +239,32 @@ async def retry_job(ctx): retry_job.__qualname__ = retry_job.__name__ - pool, worker = init_fixture_method(span_streaming, [retry_job]) + pool, worker = init_fixture_method([retry_job]) job = await pool.enqueue_job("retry_job") + items = capture_items("span") - if span_streaming: - items = capture_items("span") - - await worker.run_job(job.job_id, timestamp_ms()) - - sentry_sdk.flush() - spans = [item.payload for item in items] - - # The retry re-enqueue happens without an active span, so no producer - # (queue.submit.arq) span is created for it; only the consumer segment - # is emitted. The consumer segment is preceded by the redis spans for - # the re-enqueue, so it lands at index 2. - assert spans[2]["attributes"]["sentry.op"] == "queue.task.arq" - assert spans[2]["status"] == "ok" - assert spans[2]["name"] == "retry_job" - - await worker.run_job(job.job_id, timestamp_ms()) + await worker.run_job(job.job_id, timestamp_ms()) - sentry_sdk.flush() - spans = [item.payload for item in items] + sentry_sdk.flush() + spans = [item.payload for item in items] - assert spans[5]["attributes"]["sentry.op"] == "queue.task.arq" - assert spans[5]["status"] == "ok" - assert spans[5]["name"] == "retry_job" - else: - events = capture_events() + # The retry re-enqueue happens without an active span, so no producer + # (queue.submit.arq) span is created for it; only the consumer segment + # is emitted. The consumer segment is preceded by the redis spans for + # the re-enqueue, so it lands at index 2. + assert spans[2]["attributes"]["sentry.op"] == "queue.task.arq" + assert spans[2]["status"] == "ok" + assert spans[2]["name"] == "retry_job" - await worker.run_job(job.job_id, timestamp_ms()) - - event = events.pop(0) - assert event["contexts"]["trace"]["status"] == "aborted" - assert event["transaction"] == "retry_job" - assert event["tags"]["arq_task_id"] == job.job_id - assert event["extra"]["arq-job"]["retry"] == 1 + await worker.run_job(job.job_id, timestamp_ms()) - await worker.run_job(job.job_id, timestamp_ms()) + sentry_sdk.flush() + spans = [item.payload for item in items] - event = events.pop(0) - assert event["contexts"]["trace"]["status"] == "ok" - assert event["transaction"] == "retry_job" - assert event["tags"]["arq_task_id"] == job.job_id - assert event["extra"]["arq-job"]["retry"] == 2 + assert spans[5]["attributes"]["sentry.op"] == "queue.task.arq" + assert spans[5]["status"] == "ok" + assert spans[5]["name"] == "retry_job" @pytest.mark.parametrize( @@ -299,15 +275,12 @@ async def retry_job(ctx): "init_arq_settings", ["init_arq", "init_arq_with_dict_settings"] ) @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) -async def test_job_transaction( - capture_events, +async def test_worker_jobs( capture_items, init_arq_settings, source, job_fails, request, - span_streaming, ): async def division(_, a, b=0): return a / b @@ -323,138 +296,56 @@ async def division(_, a, b=0): functions_key, cron_jobs_key = source pool, worker = init_fixture_method( - span_streaming, **{functions_key: [division], cron_jobs_key: [cron_job]} + **{functions_key: [division], cron_jobs_key: [cron_job]} ) job = await pool.enqueue_job("division", 1, b=int(not job_fails)) + items = capture_items("event", "span") - if span_streaming: - items = capture_items("event", "span") - - await worker.run_job(job.job_id, timestamp_ms()) - - loop = asyncio.get_event_loop() - task = loop.create_task(worker.async_run()) - await asyncio.sleep(1) - - task.cancel() - - await worker.close() - - events = [item.payload for item in items if item.type == "event"] - if job_fails: - error_func_event = events.pop(0) - error_cron_event = events.pop(0) - - assert ( - error_func_event["exception"]["values"][0]["type"] - == "ZeroDivisionError" - ) - assert ( - error_func_event["exception"]["values"][0]["mechanism"]["type"] == "arq" - ) - - func_extra = error_func_event["extra"]["arq-job"] - assert func_extra["task"] == "division" - - assert ( - error_cron_event["exception"]["values"][0]["type"] - == "ZeroDivisionError" - ) - assert ( - error_cron_event["exception"]["values"][0]["mechanism"]["type"] == "arq" - ) - - cron_extra = error_cron_event["extra"]["arq-job"] - assert cron_extra["task"] == "cron:division" - - sentry_sdk.flush() - spans = [item.payload for item in items if item.type == "span"] - - task_spans = [ - span - for span in spans - if span["attributes"].get("sentry.op") == "queue.task.arq" - ] - - division_span = next(span for span in task_spans if span["name"] == "division") - assert division_span["attributes"]["sentry.segment.name.source"] == "task" - assert ( - division_span["attributes"][SPANDATA.MESSAGING_DESTINATION_NAME] - == worker.queue_name - ) - - assert any(span["name"] == "cron:division" for span in task_spans) - else: - events = capture_events() - - await worker.run_job(job.job_id, timestamp_ms()) - - loop = asyncio.get_event_loop() - task = loop.create_task(worker.async_run()) - await asyncio.sleep(1) - - task.cancel() - - await worker.close() - - if job_fails: - error_func_event = events.pop(0) - error_cron_event = events.pop(1) - - assert ( - error_func_event["exception"]["values"][0]["type"] - == "ZeroDivisionError" - ) - assert ( - error_func_event["exception"]["values"][0]["mechanism"]["type"] == "arq" - ) + await worker.run_job(job.job_id, timestamp_ms()) - func_extra = error_func_event["extra"]["arq-job"] - assert func_extra["task"] == "division" + loop = asyncio.get_event_loop() + task = loop.create_task(worker.async_run()) + await asyncio.sleep(1) - assert ( - error_cron_event["exception"]["values"][0]["type"] - == "ZeroDivisionError" - ) - assert ( - error_cron_event["exception"]["values"][0]["mechanism"]["type"] == "arq" - ) + task.cancel() - cron_extra = error_cron_event["extra"]["arq-job"] - assert cron_extra["task"] == "cron:division" + await worker.close() - [func_event, cron_event] = events + events = [item.payload for item in items if item.type == "event"] + if job_fails: + error_func_event = events.pop(0) + error_cron_event = events.pop(0) - assert func_event["type"] == "transaction" - assert func_event["transaction"] == "division" - assert func_event["transaction_info"] == {"source": "task"} - assert ( - func_event["contexts"]["trace"]["data"][SPANDATA.MESSAGING_DESTINATION_NAME] - == worker.queue_name - ) + assert error_func_event["exception"]["values"][0]["type"] == "ZeroDivisionError" + assert error_func_event["exception"]["values"][0]["mechanism"]["type"] == "arq" - assert "arq_task_id" in func_event["tags"] - assert "arq_task_retry" in func_event["tags"] + func_extra = error_func_event["extra"]["arq-job"] + assert func_extra["task"] == "division" - func_extra = func_event["extra"]["arq-job"] + assert error_cron_event["exception"]["values"][0]["type"] == "ZeroDivisionError" + assert error_cron_event["exception"]["values"][0]["mechanism"]["type"] == "arq" - assert func_extra["task"] == "division" - assert func_extra["kwargs"] == {"b": int(not job_fails)} - assert func_extra["retry"] == 1 + cron_extra = error_cron_event["extra"]["arq-job"] + assert cron_extra["task"] == "cron:division" - assert cron_event["type"] == "transaction" - assert cron_event["transaction"] == "cron:division" - assert cron_event["transaction_info"] == {"source": "task"} + sentry_sdk.flush() + spans = [item.payload for item in items if item.type == "span"] - assert "arq_task_id" in cron_event["tags"] - assert "arq_task_retry" in cron_event["tags"] + task_spans = [ + span + for span in spans + if span["attributes"].get("sentry.op") == "queue.task.arq" + ] - cron_extra = cron_event["extra"]["arq-job"] + division_span = next(span for span in task_spans if span["name"] == "division") + assert division_span["attributes"]["sentry.segment.name.source"] == "task" + assert ( + division_span["attributes"][SPANDATA.MESSAGING_DESTINATION_NAME] + == worker.queue_name + ) - assert cron_extra["task"] == "cron:division" - assert cron_extra["kwargs"] == {} - assert cron_extra["retry"] == 1 + assert any(span["name"] == "cron:division" for span in task_spans) @pytest.mark.parametrize( @@ -462,12 +353,9 @@ async def division(_, a, b=0): DATA_COLLECTION_QUEUES_CASES, ) @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_job_args_kwargs_data_collection( - capture_events, capture_items, init_arq, - span_streaming, init_kwargs, expected_args, expected_kwargs, @@ -478,22 +366,15 @@ async def division(_, a, b=1): division.__qualname__ = division.__name__ pool, worker = init_arq( - span_streaming, cls_functions=[division], init_kwargs=init_kwargs, ) job = await pool.enqueue_job("division", 1, b=0) - - if span_streaming: - items = capture_items("event") - else: - events = capture_events() + items = capture_items("event") await worker.run_job(job.job_id, timestamp_ms()) - - if span_streaming: - events = [item.payload for item in items] + events = [item.payload for item in items] (event,) = [event for event in events if "exception" in event] @@ -514,60 +395,41 @@ async def division(_, a, b=1): "init_arq_settings", ["init_arq", "init_arq_with_dict_settings"] ) @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_enqueue_job( - capture_events, capture_items, init_arq_settings, source, request, - span_streaming, ): async def dummy_job(_): pass init_fixture_method = request.getfixturevalue(init_arq_settings) - pool, _ = init_fixture_method(span_streaming, **{source: [dummy_job]}) - - if span_streaming: - items = capture_items("span") + pool, _ = init_fixture_method(**{source: [dummy_job]}) + items = capture_items("span") - with sentry_sdk.traces.start_span(name="custom parent") as span: - await pool.enqueue_job("dummy_job") + with sentry_sdk.traces.start_span(name="custom parent") as span: + await pool.enqueue_job("dummy_job") - sentry_sdk.flush() - spans = [item.payload for item in items] + sentry_sdk.flush() + spans = [item.payload for item in items] - assert spans[2]["is_segment"] is True - assert spans[2]["trace_id"] == span.trace_id - assert spans[2]["span_id"] == span.span_id + assert spans[2]["is_segment"] is True + assert spans[2]["trace_id"] == span.trace_id + assert spans[2]["span_id"] == span.span_id - assert spans[1]["attributes"]["sentry.op"] == "queue.submit.arq" - assert spans[1]["name"] == "dummy_job" - else: - events = capture_events() - - with start_transaction() as transaction: - await pool.enqueue_job("dummy_job") - - (event,) = events - - assert event["contexts"]["trace"]["trace_id"] == transaction.trace_id - assert event["contexts"]["trace"]["span_id"] == transaction.span_id - - assert len(event["spans"]) - assert event["spans"][0]["op"] == "queue.submit.arq" - assert event["spans"][0]["description"] == "dummy_job" + assert spans[1]["attributes"]["sentry.op"] == "queue.submit.arq" + assert spans[1]["name"] == "dummy_job" @pytest.mark.asyncio @pytest.mark.parametrize( "init_arq_settings", ["init_arq", "init_arq_with_dict_settings"] ) -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_execute_job_without_integration( - init_arq_settings, request, span_streaming + init_arq_settings, + request, ): async def dummy_job(_ctx): pass @@ -576,7 +438,7 @@ async def dummy_job(_ctx): dummy_job.__qualname__ = dummy_job.__name__ - pool, worker = init_fixture_method(span_streaming, [dummy_job]) + pool, worker = init_fixture_method([dummy_job]) # remove the integration to trigger the edge case get_client().integrations.pop("arq") @@ -592,54 +454,37 @@ async def dummy_job(_ctx): "init_arq_settings", ["init_arq", "init_arq_with_dict_settings"] ) @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_span_origin_producer( - capture_events, capture_items, init_arq_settings, source, request, - span_streaming, ): async def dummy_job(_): pass init_fixture_method = request.getfixturevalue(init_arq_settings) - pool, _ = init_fixture_method(span_streaming, **{source: [dummy_job]}) - - if span_streaming: - items = capture_items("span") + pool, _ = init_fixture_method(**{source: [dummy_job]}) + items = capture_items("span") - with sentry_sdk.traces.start_span(name="custom parent"): - await pool.enqueue_job("dummy_job") + with sentry_sdk.traces.start_span(name="custom parent"): + await pool.enqueue_job("dummy_job") - sentry_sdk.flush() - spans = [item.payload for item in items] - assert spans[2]["attributes"]["sentry.origin"] == "manual" - assert spans[1]["attributes"]["sentry.origin"] == "auto.queue.arq" - else: - events = capture_events() - - with start_transaction(): - await pool.enqueue_job("dummy_job") - - (event,) = events - assert event["contexts"]["trace"]["origin"] == "manual" - assert event["spans"][0]["origin"] == "auto.queue.arq" + sentry_sdk.flush() + spans = [item.payload for item in items] + assert spans[2]["attributes"]["sentry.origin"] == "manual" + assert spans[1]["attributes"]["sentry.origin"] == "auto.queue.arq" @pytest.mark.asyncio @pytest.mark.parametrize( "init_arq_settings", ["init_arq", "init_arq_with_dict_settings"] ) -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_span_origin_consumer( - capture_events, capture_items, init_arq_settings, request, - span_streaming, ): async def job(ctx): pass @@ -648,46 +493,29 @@ async def job(ctx): job.__qualname__ = job.__name__ - pool, worker = init_fixture_method(span_streaming, [job]) - - if span_streaming: - job = await pool.enqueue_job("job") - - items = capture_items("span") - - await worker.run_job(job.job_id, timestamp_ms()) - - sentry_sdk.flush() - spans = [item.payload for item in items] - - # No producer (queue.submit.arq) span is created for the re-enqueue - # triggered by the retry, since it happens without an active span, so - # the consumer segment lands at index 2. - assert spans[2]["attributes"]["sentry.op"] == "queue.task.arq" - assert spans[2]["attributes"]["sentry.origin"] == "auto.queue.arq" - assert spans[1]["attributes"]["sentry.origin"] == "auto.db.redis" - assert spans[0]["attributes"]["sentry.origin"] == "auto.db.redis" - else: - job = await pool.enqueue_job("job") + pool, worker = init_fixture_method([job]) + job = await pool.enqueue_job("job") - events = capture_events() + items = capture_items("span") - await worker.run_job(job.job_id, timestamp_ms()) + await worker.run_job(job.job_id, timestamp_ms()) - (event,) = events + sentry_sdk.flush() + spans = [item.payload for item in items] - assert event["contexts"]["trace"]["origin"] == "auto.queue.arq" - assert event["spans"][0]["origin"] == "auto.db.redis" - assert event["spans"][1]["origin"] == "auto.db.redis" + # No producer (queue.submit.arq) span is created for the re-enqueue + # triggered by the retry, since it happens without an active span, so + # the consumer segment lands at index 2. + assert spans[2]["attributes"]["sentry.op"] == "queue.task.arq" + assert spans[2]["attributes"]["sentry.origin"] == "auto.queue.arq" + assert spans[1]["attributes"]["sentry.origin"] == "auto.db.redis" + assert spans[0]["attributes"]["sentry.origin"] == "auto.db.redis" @pytest.mark.asyncio -@pytest.mark.parametrize("span_streaming", [True, False]) async def test_job_concurrency( - capture_events, capture_items, init_arq, - span_streaming, ): """ 10 - division starts @@ -707,7 +535,7 @@ async def division(_): sleepy.__qualname__ = sleepy.__name__ division.__qualname__ = division.__name__ - pool, worker = init_arq(span_streaming, [sleepy, division]) + pool, worker = init_arq([sleepy, division]) await pool.enqueue_job( "division", _job_id="123", _defer_by=timedelta(milliseconds=10) @@ -715,35 +543,19 @@ async def division(_): await pool.enqueue_job( "sleepy", _job_id="456", _defer_by=timedelta(milliseconds=70) ) + items = capture_items("event") - if span_streaming: - items = capture_items("event") - - loop = asyncio.get_event_loop() - task = loop.create_task(worker.async_run()) - await asyncio.sleep(1) - - task.cancel() - - await worker.close() - - events = [item.payload for item in items] - exception_event = events[0] - assert exception_event["exception"]["values"][0]["type"] == "ZeroDivisionError" - assert exception_event["transaction"] == "division" - else: - events = capture_events() - - loop = asyncio.get_event_loop() - task = loop.create_task(worker.async_run()) - await asyncio.sleep(1) + loop = asyncio.get_event_loop() + task = loop.create_task(worker.async_run()) + await asyncio.sleep(1) - task.cancel() + task.cancel() - await worker.close() + await worker.close() - (exception_event,) = (event for event in events if "exception" in event) - assert exception_event["exception"]["values"][0]["type"] == "ZeroDivisionError" - assert exception_event["transaction"] == "division" + events = [item.payload for item in items] + exception_event = events[0] + assert exception_event["exception"]["values"][0]["type"] == "ZeroDivisionError" + assert exception_event["transaction"] == "division" assert exception_event["extra"]["arq-job"]["task"] == "division"