fix(dramatiq): Gate request body collection on data_collection experiment

sentry_sdk/integrations/dramatiq.py

@@ -17,6 +17,7 @@
     AnnotatedValue,
     capture_internal_exceptions,
     event_from_exception,
+    has_data_collection_enabled,
 )
 
 R = TypeVar("R")
@@ -241,10 +242,17 @@ def extract_into_event(self, event: "Event") -> None:
         request_info = contexts.setdefault("dramatiq", {})
         request_info["type"] = "dramatiq"
 
-        data: "Optional[Union[AnnotatedValue, Dict[str, Any]]]" = None
-        if not request_body_within_bounds(client, self.content_length()):
-            data = AnnotatedValue.removed_because_over_size_limit()
-        else:
-            data = self.message_data
+        attach_request_body = True
+        if has_data_collection_enabled(client.options):
+            attach_request_body = (
+                "incoming_request" in client.options["data_collection"]["http_bodies"]
+            )
+
+        if attach_request_body:
+            data: "Optional[Union[AnnotatedValue, Dict[str, Any]]]" = None
+            if not request_body_within_bounds(client, self.content_length()):
+                data = AnnotatedValue.removed_because_over_size_limit()
+            else:
+                data = self.message_data
 
-        request_info["data"] = data
+            request_info["data"] = data

tests/integrations/dramatiq/test_dramatiq.py

@@ -455,6 +455,75 @@
     assert isinstance(request_data["message_timestamp"], int)
 
 
+@pytest.mark.parametrize(
+    "broker,expect_message_data",
+    [
+        pytest.param({}, True, id="data_collection_not_enabled"),
+        pytest.param(
+            {
+                "_experiments": {
+                    "data_collection": {"http_bodies": ["incoming_request"]}
+                }
+            },
+            True,
+            id="data_collection_http_bodies_incoming_request",
+        ),
+        pytest.param(
+            {"_experiments": {"data_collection": {"http_bodies": []}}},
+            False,
+            id="data_collection_http_bodies_empty",
+        ),
+    ],
+    indirect=["broker"],
+)
+def test_that_message_data_is_gated_by_data_collection(
+    broker, worker, capture_events, expect_message_data
+):
+    events = capture_events()
+
+    @dramatiq.actor(max_retries=0)
+    def dummy_actor(x, y):
+        return x / y
+
+    dummy_actor.send_with_options(args=(1, 0), max_retries=0)
+    broker.join(dummy_actor.queue_name, fail_fast=False)
+    worker.join()
+
+    (event,) = events
+    dramatiq_context = event["contexts"]["dramatiq"]
+
+    if expect_message_data:
+        assert dramatiq_context["data"]["actor_name"] == "dummy_actor"
+        assert dramatiq_context["data"]["args"] == [1, 0]
+    else:
+        assert "data" not in dramatiq_context
+
+
+@pytest.mark.parametrize(
+    "broker",
+    [{"_experiments": {"data_collection": {"http_bodies": []}}}],
+    indirect=True,
+)
+def test_that_dramatiq_context_type_is_set_regardless_of_data_collection(
+    broker, worker, capture_events
+):
+    events = capture_events()
+
+    @dramatiq.actor(max_retries=0)
+    def dummy_actor(x, y):
+        return x / y
+
+    dummy_actor.send_with_options(args=(1, 0), max_retries=0)
+    broker.join(dummy_actor.queue_name, fail_fast=False)
+    worker.join()
+
+    (event,) = events
+    dramatiq_context = event["contexts"]["dramatiq"]
+
+    assert dramatiq_context["type"] == "dramatiq"
+    assert "data" not in dramatiq_context
+
+
 @pytest.mark.parametrize(
     "fail_fast",
     [