feat(alerts): Set up prune_old_open_period_activity (#112243)

Our backlog is only 17m or so rows, but by reusing a task pattern we've
already used before we should be able to safely get that quickly cleaned
up in all regions.

A job or post-deploy migration would also be an option, but this is no
harder and more or less guarantees that next time we look it'll be dealt
with.

Author: kcons

src/sentry/options/defaults.py

@@ -3575,18 +3575,18 @@
     default=10000,
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
-# Tuning knobs for the periodic fire-history cleanup task.
+# Tuning knobs for the periodic open-period-activity cleanup task.
 # time_limit is a wall-clock budget checked *between* batches, so a single
 # batch that exceeds it will still run to completion. Setting it to 0
 # prevents any batches from running.
 register(
-    "workflow_engine.fire_history_cleanup.time_limit_seconds",
+    "workflow_engine.open_period_activity_cleanup.time_limit_seconds",
     type=Float,
     default=5.0,
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
 register(
-    "workflow_engine.fire_history_cleanup.batch_size",
+    "workflow_engine.open_period_activity_cleanup.batch_size",
     type=Int,
     default=10000,
     flags=FLAG_AUTOMATOR_MODIFIABLE,

src/sentry/workflow_engine/tasks/__init__.py

@@ -2,9 +2,9 @@
     "process_delayed_workflows",
     "process_workflow_activity",
     "process_workflows_event",
-    "prune_old_fire_history",
+    "prune_old_open_period_activity",
 ]
 
-from .cleanup import prune_old_fire_history
+from .cleanup import prune_old_open_period_activity
 from .delayed_workflows import process_delayed_workflows
 from .workflows import process_workflow_activity, process_workflows_event

src/sentry/workflow_engine/tasks/cleanup.py

@@ -15,28 +15,30 @@
 
 logger = logging.getLogger(__name__)
 
-FIRE_HISTORY_RETENTION_DAYS = 90
+OPEN_PERIOD_ACTIVITY_RETENTION_DAYS = 90
 
 
 @instrumented_task(
-    name="sentry.workflow_engine.tasks.cleanup.prune_old_fire_history",
+    name="sentry.workflow_engine.tasks.cleanup.prune_old_open_period_activity",
     namespace=workflow_engine_tasks,
     processing_deadline_duration=15,
     silo_mode=SiloMode.CELL,
 )
-def prune_old_fire_history() -> None:
-    from sentry.workflow_engine.models import WorkflowFireHistory
+def prune_old_open_period_activity() -> None:
+    from sentry.models.groupopenperiodactivity import GroupOpenPeriodActivity
 
-    time_limit: float = options.get("workflow_engine.fire_history_cleanup.time_limit_seconds")
-    batch_size: int = options.get("workflow_engine.fire_history_cleanup.batch_size")
+    time_limit: float = options.get(
+        "workflow_engine.open_period_activity_cleanup.time_limit_seconds"
+    )
+    batch_size: int = options.get("workflow_engine.open_period_activity_cleanup.batch_size")
 
-    cutoff = timezone.now() - timedelta(days=FIRE_HISTORY_RETENTION_DAYS)
+    cutoff = timezone.now() - timedelta(days=OPEN_PERIOD_ACTIVITY_RETENTION_DAYS)
     start = time.time()
     batches_deleted = 0
 
     while (time.time() - start) < time_limit:
         has_more = bulk_delete_objects(
-            WorkflowFireHistory,
+            GroupOpenPeriodActivity,
             limit=batch_size,
             logger=logger,
             date_added__lte=cutoff,
@@ -46,7 +48,7 @@ def prune_old_fire_history() -> None:
         batches_deleted += 1
 
     metrics.incr(
-        "workflow_engine.tasks.prune_old_fire_history.batches_deleted",
+        "workflow_engine.tasks.prune_old_open_period_activity.batches_deleted",
         amount=batches_deleted,
         sample_rate=1.0,
     )

tests/sentry/workflow_engine/tasks/test_cleanup.py

@@ -3,76 +3,80 @@
 
 from django.utils import timezone
 
+from sentry.models.groupopenperiod import GroupOpenPeriod
+from sentry.models.groupopenperiodactivity import GroupOpenPeriodActivity, OpenPeriodActivityType
 from sentry.testutils.cases import TestCase
 from sentry.testutils.helpers.options import override_options
 from sentry.utils.query import bulk_delete_objects
-from sentry.workflow_engine.models import WorkflowFireHistory
 from sentry.workflow_engine.tasks.cleanup import (
-    FIRE_HISTORY_RETENTION_DAYS,
-    prune_old_fire_history,
+    OPEN_PERIOD_ACTIVITY_RETENTION_DAYS,
+    prune_old_open_period_activity,
 )
 
 
-class TestPruneOldFireHistory(TestCase):
-    def _create_fire_history(self, days_ago: int) -> WorkflowFireHistory:
-        workflow = self.create_workflow(organization=self.organization)
+class TestPruneOldOpenPeriodActivity(TestCase):
+    def _create_activity(self, days_ago: int) -> GroupOpenPeriodActivity:
         group = self.create_group(project=self.project)
-        obj = WorkflowFireHistory.objects.create(workflow=workflow, group=group, event_id="abc123")
+        open_period = GroupOpenPeriod.objects.get(group=group)
+        obj = GroupOpenPeriodActivity.objects.create(
+            group_open_period=open_period,
+            type=OpenPeriodActivityType.OPENED,
+        )
         if days_ago > 0:
             backdated = timezone.now() - timedelta(days=days_ago)
-            WorkflowFireHistory.objects.filter(id=obj.id).update(date_added=backdated)
+            GroupOpenPeriodActivity.objects.filter(id=obj.id).update(date_added=backdated)
             obj.refresh_from_db()
         return obj
 
     def test_noop_when_nothing_to_delete(self) -> None:
         with patch("sentry.workflow_engine.tasks.cleanup.metrics") as mock_metrics:
-            prune_old_fire_history()
+            prune_old_open_period_activity()
 
         mock_metrics.incr.assert_called_once_with(
-            "workflow_engine.tasks.prune_old_fire_history.batches_deleted",
+            "workflow_engine.tasks.prune_old_open_period_activity.batches_deleted",
             amount=0,
             sample_rate=1.0,
         )
 
     def test_deletes_rows_older_than_retention(self) -> None:
-        old = self._create_fire_history(days_ago=FIRE_HISTORY_RETENTION_DAYS + 1)
-        recent = self._create_fire_history(days_ago=FIRE_HISTORY_RETENTION_DAYS - 1)
+        old = self._create_activity(days_ago=OPEN_PERIOD_ACTIVITY_RETENTION_DAYS + 1)
+        recent = self._create_activity(days_ago=OPEN_PERIOD_ACTIVITY_RETENTION_DAYS - 1)
 
-        prune_old_fire_history()
+        prune_old_open_period_activity()
 
-        assert not WorkflowFireHistory.objects.filter(id=old.id).exists()
-        assert WorkflowFireHistory.objects.filter(id=recent.id).exists()
+        assert not GroupOpenPeriodActivity.objects.filter(id=old.id).exists()
+        assert GroupOpenPeriodActivity.objects.filter(id=recent.id).exists()
 
     def test_preserves_recent_rows(self) -> None:
         rows = [
-            self._create_fire_history(days_ago=1),
-            self._create_fire_history(days_ago=30),
-            self._create_fire_history(days_ago=FIRE_HISTORY_RETENTION_DAYS - 1),
+            self._create_activity(days_ago=1),
+            self._create_activity(days_ago=30),
+            self._create_activity(days_ago=OPEN_PERIOD_ACTIVITY_RETENTION_DAYS - 1),
         ]
 
-        prune_old_fire_history()
+        prune_old_open_period_activity()
 
         for row in rows:
-            assert WorkflowFireHistory.objects.filter(id=row.id).exists()
+            assert GroupOpenPeriodActivity.objects.filter(id=row.id).exists()
 
-    @override_options({"workflow_engine.fire_history_cleanup.batch_size": 10})
+    @override_options({"workflow_engine.open_period_activity_cleanup.batch_size": 10})
     def test_multiple_batches(self) -> None:
         old_ids = []
         for _ in range(25):
-            obj = self._create_fire_history(days_ago=FIRE_HISTORY_RETENTION_DAYS + 1)
+            obj = self._create_activity(days_ago=OPEN_PERIOD_ACTIVITY_RETENTION_DAYS + 1)
             old_ids.append(obj.id)
 
         with patch("sentry.workflow_engine.tasks.cleanup.metrics") as mock_metrics:
-            prune_old_fire_history()
+            prune_old_open_period_activity()
 
-        assert WorkflowFireHistory.objects.filter(id__in=old_ids).count() == 0
+        assert GroupOpenPeriodActivity.objects.filter(id__in=old_ids).count() == 0
         mock_metrics.incr.assert_called_once()
         assert mock_metrics.incr.call_args.kwargs["amount"] >= 2
 
     @override_options(
         {
-            "workflow_engine.fire_history_cleanup.batch_size": 10,
-            "workflow_engine.fire_history_cleanup.time_limit_seconds": 5.0,
+            "workflow_engine.open_period_activity_cleanup.batch_size": 10,
+            "workflow_engine.open_period_activity_cleanup.time_limit_seconds": 5.0,
         }
     )
     @patch("sentry.workflow_engine.tasks.cleanup.time")
@@ -82,41 +86,41 @@ def test_time_bounded_leaves_remaining_rows(self, mock_time: MagicMock) -> None:
 
         old_ids = []
         for _ in range(25):
-            obj = self._create_fire_history(days_ago=FIRE_HISTORY_RETENTION_DAYS + 1)
+            obj = self._create_activity(days_ago=OPEN_PERIOD_ACTIVITY_RETENTION_DAYS + 1)
             old_ids.append(obj.id)
 
         with patch("sentry.workflow_engine.tasks.cleanup.metrics"):
-            prune_old_fire_history()
+            prune_old_open_period_activity()
 
-        remaining = WorkflowFireHistory.objects.filter(id__in=old_ids).count()
+        remaining = GroupOpenPeriodActivity.objects.filter(id__in=old_ids).count()
         assert remaining == 15  # only one batch of 10 was deleted
 
-    @override_options({"workflow_engine.fire_history_cleanup.batch_size": 5})
+    @override_options({"workflow_engine.open_period_activity_cleanup.batch_size": 5})
     def test_options_honored(self) -> None:
         old_ids = []
         for _ in range(12):
-            obj = self._create_fire_history(days_ago=FIRE_HISTORY_RETENTION_DAYS + 1)
+            obj = self._create_activity(days_ago=OPEN_PERIOD_ACTIVITY_RETENTION_DAYS + 1)
             old_ids.append(obj.id)
 
         with patch(
             "sentry.workflow_engine.tasks.cleanup.bulk_delete_objects",
             wraps=bulk_delete_objects,
         ) as spy:
-            prune_old_fire_history()
+            prune_old_open_period_activity()
 
         for call in spy.call_args_list:
             assert call.kwargs["limit"] == 5
 
-        assert WorkflowFireHistory.objects.filter(id__in=old_ids).count() == 0
+        assert GroupOpenPeriodActivity.objects.filter(id__in=old_ids).count() == 0
 
     def test_metrics_emitted(self) -> None:
-        self._create_fire_history(days_ago=FIRE_HISTORY_RETENTION_DAYS + 1)
+        self._create_activity(days_ago=OPEN_PERIOD_ACTIVITY_RETENTION_DAYS + 1)
 
         with patch("sentry.workflow_engine.tasks.cleanup.metrics") as mock_metrics:
-            prune_old_fire_history()
+            prune_old_open_period_activity()
 
         mock_metrics.incr.assert_called_once_with(
-            "workflow_engine.tasks.prune_old_fire_history.batches_deleted",
+            "workflow_engine.tasks.prune_old_open_period_activity.batches_deleted",
             amount=1,  # 1 row fits in a single batch
             sample_rate=1.0,
         )