fix(aci): Don't propagate traces to buffer processing tasks (#93521)

We don't have any real need for parent or peer traces in these tasks,
and not propagating should leave us with dramatically smaller traces
that will load much more quickly and be easier to read.

Note that this affects delayed_workflow and delayed_processing;
delayed_workflow is the one we've had more issues loading, but I don't
see it as having significant downside for either so I didn't make it
conditional by task.

Author: kcons

src/sentry/rules/processing/buffer_processing.py

@@ -107,7 +107,9 @@ def process_in_batches(project_id: int, processing_type: str) -> None:
     metrics.distribution(f"{processing_type}.event_count", event_count)
 
     if event_count < batch_size:
-        return task.delay(project_id)
+        return task.apply_async(
+            kwargs={"project_id": project_id}, headers={"sentry-propagate-traces": False}
+        )
 
     if should_emit_logs:
         logger.info(
@@ -133,7 +135,10 @@ def process_in_batches(project_id: int, processing_type: str) -> None:
             # remove the batched items from the project alertgroup_to_event_data
             buffer.backend.delete_hash(**asdict(hash_args), fields=list(batch.keys()))
 
-            task.delay(project_id, batch_key)
+            task.apply_async(
+                kwargs={"project_id": project_id, "batch_key": batch_key},
+                headers={"sentry-propagate-traces": False},
+            )
 
 
 def process_buffer() -> None:

tests/sentry/rules/processing/test_buffer_processing.py

@@ -253,22 +253,26 @@ def setUp(self):
         self.group_three = self.create_group(self.project)
         self.rule = self.create_alert_rule()
 
-    @patch("sentry.rules.processing.delayed_processing.apply_delayed.delay")
+    @patch("sentry.rules.processing.delayed_processing.apply_delayed.apply_async")
     def test_no_redis_data(self, mock_apply_delayed):
         process_in_batches(self.project.id, "delayed_processing")
-        mock_apply_delayed.assert_called_once_with(self.project.id)
+        mock_apply_delayed.assert_called_once_with(
+            kwargs={"project_id": self.project.id}, headers={"sentry-propagate-traces": False}
+        )
 
-    @patch("sentry.rules.processing.delayed_processing.apply_delayed.delay")
+    @patch("sentry.rules.processing.delayed_processing.apply_delayed.apply_async")
     def test_basic(self, mock_apply_delayed):
         self.push_to_hash(self.project.id, self.rule.id, self.group.id)
         self.push_to_hash(self.project.id, self.rule.id, self.group_two.id)
         self.push_to_hash(self.project.id, self.rule.id, self.group_three.id)
 
         process_in_batches(self.project.id, "delayed_processing")
-        mock_apply_delayed.assert_called_once_with(self.project.id)
+        mock_apply_delayed.assert_called_once_with(
+            kwargs={"project_id": self.project.id}, headers={"sentry-propagate-traces": False}
+        )
 
     @override_options({"delayed_processing.batch_size": 2})
-    @patch("sentry.rules.processing.delayed_processing.apply_delayed.delay")
+    @patch("sentry.rules.processing.delayed_processing.apply_delayed.apply_async")
     def test_batch(self, mock_apply_delayed):
         self.push_to_hash(self.project.id, self.rule.id, self.group.id)
         self.push_to_hash(self.project.id, self.rule.id, self.group_two.id)
@@ -278,11 +282,11 @@ def test_batch(self, mock_apply_delayed):
         assert mock_apply_delayed.call_count == 2
 
         # Validate the batches are created correctly
-        batch_one_key = mock_apply_delayed.call_args_list[0][0][1]
+        batch_one_key = mock_apply_delayed.call_args_list[0][1]["kwargs"]["batch_key"]
         batch_one = buffer.backend.get_hash(
             model=Project, field={"project_id": self.project.id, "batch_key": batch_one_key}
         )
-        batch_two_key = mock_apply_delayed.call_args_list[1][0][1]
+        batch_two_key = mock_apply_delayed.call_args_list[1][1]["kwargs"]["batch_key"]
         batch_two = buffer.backend.get_hash(
             model=Project, field={"project_id": self.project.id, "batch_key": batch_two_key}
         )

tests/sentry/rules/processing/test_delayed_processing.py

@@ -1407,7 +1407,7 @@ def test_cleanup_redis(self):
         assert rule_group_data == {}
 
     @override_options({"delayed_processing.batch_size": 2})
-    @patch("sentry.rules.processing.delayed_processing.apply_delayed.delay")
+    @patch("sentry.rules.processing.delayed_processing.apply_delayed.apply_async")
     def test_batched_cleanup(self, mock_apply_delayed):
         group_two = self.create_group(self.project)
         group_three = self.create_group(self.project)
@@ -1422,8 +1422,8 @@ def test_batched_cleanup(self, mock_apply_delayed):
         rules_to_groups[self.rule.id].add(group_three.id)
 
         process_in_batches(self.project.id, "delayed_processing")
-        batch_one_key = mock_apply_delayed.call_args_list[0][0][1]
-        batch_two_key = mock_apply_delayed.call_args_list[1][0][1]
+        batch_one_key = mock_apply_delayed.call_args_list[0][1]["kwargs"]["batch_key"]
+        batch_two_key = mock_apply_delayed.call_args_list[1][1]["kwargs"]["batch_key"]
 
         # Verify process_rulegroups_in_batches removed the data from the buffer
         rule_group_data = buffer.backend.get_hash(Project, {"project_id": self.project.id})

tests/sentry/workflow_engine/processors/test_delayed_workflow.py

@@ -881,14 +881,16 @@ def test_cleanup_redis(self):
         assert data == {}
 
     @override_options({"delayed_processing.batch_size": 2})
-    @patch("sentry.workflow_engine.processors.delayed_workflow.process_delayed_workflows.delay")
+    @patch(
+        "sentry.workflow_engine.processors.delayed_workflow.process_delayed_workflows.apply_async"
+    )
     def test_batched_cleanup(self, mock_process_delayed):
         self._push_base_events()
         all_data = buffer.backend.get_hash(Workflow, {"project_id": self.project.id})
 
         process_in_batches(self.project.id, "delayed_workflow")
-        batch_one_key = mock_process_delayed.call_args_list[0][0][1]
-        batch_two_key = mock_process_delayed.call_args_list[1][0][1]
+        batch_one_key = mock_process_delayed.call_args_list[0][1]["kwargs"]["batch_key"]
+        batch_two_key = mock_process_delayed.call_args_list[1][1]["kwargs"]["batch_key"]
 
         # Verify we removed the data from the buffer
         data = buffer.backend.get_hash(Workflow, {"project_id": self.project.id})