feat(issue-alert): Create initial delayed processor function (#68371)

Create the intial `process_delayed_alert_conditions` method that fetches
data from the redis buffer, iterates through each project, and checks
the relevant issue alerts and fires it's actions if needed.

I also moved the `processor.py` file into a folder with the new delayed
file.

Closes
https://github.com/getsentry/team-core-product-foundations/issues/258

Author: schew2381

src/sentry/api/endpoints/project_rule_actions.py

@@ -8,7 +8,7 @@
 from sentry.api.bases import ProjectAlertRulePermission, ProjectEndpoint
 from sentry.api.serializers.rest_framework import RuleActionSerializer
 from sentry.models.rule import Rule
-from sentry.rules.processor import RuleProcessor
+from sentry.rules.processing.processor import RuleProcessor
 from sentry.utils.safe import safe_execute
 from sentry.utils.samples import create_sample_event
 

src/sentry/api/endpoints/project_rules.py

@@ -32,7 +32,7 @@
 from sentry.models.user import User
 from sentry.rules.actions import trigger_sentry_app_action_creators_for_issues
 from sentry.rules.actions.base import instantiate_action
-from sentry.rules.processor import is_condition_slow
+from sentry.rules.processing.processor import is_condition_slow
 from sentry.signals import alert_rule_created
 from sentry.tasks.integrations.slack import find_channel_id_for_rule
 from sentry.utils import metrics

src/sentry/rules/history/preview.py

@@ -18,7 +18,7 @@
     get_update_kwargs_for_group,
     get_update_kwargs_for_groups,
 )
-from sentry.rules.processor import get_match_function
+from sentry.rules.processing.processor import get_match_function
 from sentry.snuba.dataset import Dataset
 from sentry.types.condition_activity import (
     FREQUENCY_CONDITION_BUCKET_SIZE,

src/sentry/rules/processing/__init__.py

@@ -0,0 +1,34 @@
+import logging
+from collections.abc import Mapping
+
+from sentry.buffer.redis import BufferHookEvent, RedisBuffer, redis_buffer_registry
+from sentry.models.project import Project
+from sentry.utils import metrics
+from sentry.utils.query import RangeQuerySetWrapper
+from sentry.utils.safe import safe_execute
+
+logger = logging.getLogger("sentry.rules.delayed_processing")
+
+
+PROJECT_ID_BUFFER_LIST_KEY = "project_id_buffer_list"
+
+
+@redis_buffer_registry.add_handler(BufferHookEvent.FLUSH)
+def process_delayed_alert_conditions(buffer: RedisBuffer) -> None:
+    with metrics.timer("delayed_processing.process_all_conditions.duration"):
+        project_ids = buffer.get_set(PROJECT_ID_BUFFER_LIST_KEY)
+
+        for project in RangeQuerySetWrapper(Project.objects.filter(id__in=project_ids)):
+            rulegroup_event_mapping = buffer.get_hash(model=Project, field={"id": project.id})
+
+            with metrics.timer("delayed_processing.process_project.duration"):
+                safe_execute(
+                    apply_delayed,
+                    project,
+                    rulegroup_event_mapping,
+                    _with_transaction=False,
+                )
+
+
+def apply_delayed(project: Project, rule_group_pairs: Mapping[str, str]) -> None:
+    pass

src/sentry/rules/processing/delayed_processing.py

@@ -1091,7 +1091,7 @@ def process_rules(job: PostProcessJob) -> None:
     if job["is_reprocessed"]:
         return
 
-    from sentry.rules.processor import RuleProcessor
+    from sentry.rules.processing.processor import RuleProcessor
 
     group_event = job["event"]
     if isinstance(group_event, Event):

src/sentry/rules/processor.py renamed to src/sentry/rules/processing/processor.py

@@ -11,6 +11,7 @@
 from sentry.buffer.redis import BufferHookEvent, RedisBuffer, redis_buffer_registry
 from sentry.models.group import Group
 from sentry.models.project import Project
+from sentry.rules.processing.delayed_processing import PROJECT_ID_BUFFER_LIST_KEY
 from sentry.testutils.helpers.datetime import freeze_time
 from sentry.testutils.pytest.fixtures import django_db_all
 from sentry.utils import json
@@ -268,7 +269,6 @@ def group_rule_data_by_project_id(self, buffer, project_ids):
         return project_ids_to_rule_data
 
     def test_enqueue(self):
-        PROJECT_ID_BUFFER_LIST_KEY = "project_id_buffer_list"
         project_id = 1
         rule_id = 2
         group_id = 3

src/sentry/tasks/post_process.py

@@ -0,0 +1,53 @@
+from unittest.mock import Mock, patch
+
+from sentry.db import models
+from sentry.rules.processing.delayed_processing import (
+    apply_delayed,
+    process_delayed_alert_conditions,
+)
+from sentry.testutils.cases import TestCase
+
+
+class ProcessDelayedAlertConditionsTest(TestCase):
+    def get_rulegroup_event_mapping_from_input(
+        self, model: type[models.Model], field: dict[str, models.Model | str | int]
+    ):
+        # There will only be one event per rulegroup
+        proj_id = field.popitem()[1]
+        return self.buffer_mapping[proj_id]
+
+    @patch("sentry.rules.processing.delayed_processing.safe_execute")
+    def test_fetches_from_buffer_and_executes(self, mock_safe_execute):
+        project_two = self.create_project()
+
+        rulegroup_event_mapping_one = {
+            f"{self.project.id}:1": "event_1",
+            f"{project_two.id}:2": "event_2",
+        }
+        rulegroup_event_mapping_two = {
+            f"{self.project.id}:3": "event_3",
+            f"{project_two.id}:4": "event_4",
+        }
+        self.buffer_mapping = {
+            self.project.id: rulegroup_event_mapping_one,
+            project_two.id: rulegroup_event_mapping_two,
+        }
+
+        mock_buffer = Mock()
+        mock_buffer.get_set.return_value = self.buffer_mapping.keys()
+        # To get the correct mapping, we need to return the correct
+        # rulegroup_event mapping based on the project_id input
+        mock_buffer.get_hash.side_effect = self.get_rulegroup_event_mapping_from_input
+
+        process_delayed_alert_conditions(mock_buffer)
+
+        for project, rule_group_event_mapping in (
+            (self.project, rulegroup_event_mapping_one),
+            (project_two, rulegroup_event_mapping_two),
+        ):
+            mock_safe_execute.assert_any_call(
+                apply_delayed,
+                project,
+                rule_group_event_mapping,
+                _with_transaction=False,
+            )

tests/sentry/buffer/test_redis.py

@@ -17,7 +17,7 @@
 from sentry.rules import init_registry
 from sentry.rules.conditions import EventCondition
 from sentry.rules.filters.base import EventFilter
-from sentry.rules.processor import RuleProcessor
+from sentry.rules.processing.processor import RuleProcessor
 from sentry.testutils.cases import TestCase
 from sentry.testutils.helpers import install_slack
 from sentry.testutils.helpers.features import with_feature
@@ -37,7 +37,7 @@
 
 
 class MockConditionTrue(EventCondition):
-    id = "tests.sentry.rules.test_processor.MockConditionTrue"
+    id = "tests.sentry.rules.processing.test_processor.MockConditionTrue"
     label = "Mock condition which always passes."
 
     def passes(self, event, state):
@@ -317,7 +317,9 @@ def test_multiple_rules(self):
 
         # Test that we don't get errors if we try to create statuses that already exist due to a
         # race condition
-        with mock.patch("sentry.rules.processor.GroupRuleStatus") as mocked_GroupRuleStatus:
+        with mock.patch(
+            "sentry.rules.processing.processor.GroupRuleStatus"
+        ) as mocked_GroupRuleStatus:
             call_count = 0
 
             def mock_filter(*args, **kwargs):
@@ -339,7 +341,7 @@ def mock_filter(*args, **kwargs):
         [
             "sentry.mail.actions.NotifyEmailAction",
             "sentry.rules.conditions.event_frequency.EventFrequencyCondition",
-            "tests.sentry.rules.test_processor.MockConditionTrue",
+            "tests.sentry.rules.processing.test_processor.MockConditionTrue",
         ],
     )
     def test_slow_conditions_evaluate_last(self):
@@ -349,14 +351,14 @@ def test_slow_conditions_evaluate_last(self):
             data={
                 "conditions": [
                     {"id": "sentry.rules.conditions.event_frequency.EventFrequencyCondition"},
-                    {"id": "tests.sentry.rules.test_processor.MockConditionTrue"},
+                    {"id": "tests.sentry.rules.processing.test_processor.MockConditionTrue"},
                 ],
                 "action_match": "any",
                 "actions": [EMAIL_ACTION_DATA],
             },
         )
         with (
-            patch("sentry.rules.processor.rules", init_registry()),
+            patch("sentry.rules.processing.processor.rules", init_registry()),
             patch(
                 "sentry.rules.conditions.event_frequency.BaseEventFrequencyCondition.passes"
             ) as passes,
@@ -376,15 +378,15 @@ def test_slow_conditions_evaluate_last(self):
 
 
 class MockFilterTrue(EventFilter):
-    id = "tests.sentry.rules.test_processor.MockFilterTrue"
+    id = "tests.sentry.rules.processing.test_processor.MockFilterTrue"
     label = "Mock filter which always passes."
 
     def passes(self, event, state):
         return True
 
 
 class MockFilterFalse(EventFilter):
-    id = "tests.sentry.rules.test_processor.MockFilterFalse"
+    id = "tests.sentry.rules.processing.test_processor.MockFilterFalse"
     label = "Mock filter which never passes."
 
     def passes(self, event, state):
@@ -395,8 +397,8 @@ class RuleProcessorTestFilters(TestCase):
     MOCK_SENTRY_RULES_WITH_FILTERS = (
         "sentry.mail.actions.NotifyEmailAction",
         "sentry.rules.conditions.every_event.EveryEventCondition",
-        "tests.sentry.rules.test_processor.MockFilterTrue",
-        "tests.sentry.rules.test_processor.MockFilterFalse",
+        "tests.sentry.rules.processing.test_processor.MockFilterTrue",
+        "tests.sentry.rules.processing.test_processor.MockFilterFalse",
     )
 
     def setUp(self):
@@ -406,7 +408,7 @@ def setUp(self):
     @patch("sentry.constants._SENTRY_RULES", MOCK_SENTRY_RULES_WITH_FILTERS)
     def test_filter_passes(self):
         # setup a simple alert rule with 1 condition and 1 filter that always pass
-        filter_data = {"id": "tests.sentry.rules.test_processor.MockFilterTrue"}
+        filter_data = {"id": "tests.sentry.rules.processing.test_processor.MockFilterTrue"}
 
         Rule.objects.filter(project=self.group_event.project).delete()
         ProjectOwnership.objects.create(project_id=self.project.id, fallthrough=True)
@@ -418,7 +420,7 @@ def test_filter_passes(self):
             },
         )
         # patch the rule registry to contain the mocked rules
-        with patch("sentry.rules.processor.rules", init_registry()):
+        with patch("sentry.rules.processing.processor.rules", init_registry()):
             rp = RuleProcessor(
                 self.group_event,
                 is_new=True,
@@ -436,7 +438,7 @@ def test_filter_passes(self):
     @patch("sentry.constants._SENTRY_RULES", MOCK_SENTRY_RULES_WITH_FILTERS)
     def test_filter_fails(self):
         # setup a simple alert rule with 1 condition and 1 filter that doesn't pass
-        filter_data = {"id": "tests.sentry.rules.test_processor.MockFilterFalse"}
+        filter_data = {"id": "tests.sentry.rules.processing.test_processor.MockFilterFalse"}
 
         Rule.objects.filter(project=self.group_event.project).delete()
         self.rule = Rule.objects.create(
@@ -447,7 +449,7 @@ def test_filter_fails(self):
             },
         )
         # patch the rule registry to contain the mocked rules
-        with patch("sentry.rules.processor.rules", init_registry()):
+        with patch("sentry.rules.processing.processor.rules", init_registry()):
             rp = RuleProcessor(
                 self.group_event,
                 is_new=True,
@@ -553,15 +555,15 @@ def test_last_active_too_recent(self):
         )
 
         with mock.patch(
-            "sentry.rules.processor.RuleProcessor.bulk_get_rule_status",
+            "sentry.rules.processing.processor.RuleProcessor.bulk_get_rule_status",
             return_value={self.rule.id: grs},
         ):
             results = list(rp.apply())
             assert len(results) == 0
 
-    @mock.patch("sentry.rules.processor.RuleProcessor.logger")
+    @mock.patch("sentry.rules.processing.processor.RuleProcessor.logger")
     def test_invalid_predicate(self, mock_logger):
-        filter_data = {"id": "tests.sentry.rules.test_processor.MockFilterTrue"}
+        filter_data = {"id": "tests.sentry.rules.processing.test_processor.MockFilterTrue"}
 
         Rule.objects.filter(project=self.group_event.project).delete()
         ProjectOwnership.objects.create(project_id=self.project.id, fallthrough=True)
@@ -573,7 +575,7 @@ def test_invalid_predicate(self, mock_logger):
             },
         )
 
-        with patch("sentry.rules.processor.get_match_function", return_value=None):
+        with patch("sentry.rules.processing.processor.get_match_function", return_value=None):
             rp = RuleProcessor(
                 self.group_event,
                 is_new=True,