feat(aci): set up issue stream detector (#102280)

Author: cathteng

src/sentry/projects/project_rules/creator.py

@@ -10,10 +10,8 @@
 from sentry.models.project import Project
 from sentry.models.rule import Rule, RuleSource
 from sentry.types.actor import Actor
-from sentry.workflow_engine.migration_helpers.issue_alert_migration import (
-    IssueAlertMigrator,
-    ensure_default_error_detector,
-)
+from sentry.workflow_engine.migration_helpers.issue_alert_migration import IssueAlertMigrator
+from sentry.workflow_engine.processors.detector import ensure_default_detectors
 
 logger = logging.getLogger(__name__)
 
@@ -36,7 +34,7 @@ def run(self) -> Rule:
         if features.has(
             "organizations:workflow-engine-issue-alert-dual-write", self.project.organization
         ):
-            ensure_default_error_detector(self.project)
+            ensure_default_detectors(self.project)
 
         with transaction.atomic(router.db_for_write(Rule)):
             self.rule = self._create_rule()

src/sentry/workflow_engine/migration_helpers/issue_alert_migration.py

@@ -1,20 +1,13 @@
 import logging
 from typing import Any
 
-from django.db import router, transaction
-from rest_framework import status
-
-from sentry.api.exceptions import SentryAPIException
 from sentry.constants import ObjectStatus
 from sentry.grouping.grouptype import ErrorGroupType
-from sentry.locks import locks
-from sentry.models.project import Project
 from sentry.models.rule import Rule, RuleSource
 from sentry.models.rulesnooze import RuleSnooze
 from sentry.rules.conditions.event_frequency import EventUniqueUserFrequencyConditionWithConditions
 from sentry.rules.conditions.every_event import EveryEventCondition
 from sentry.rules.processing.processor import split_conditions_and_filters
-from sentry.utils.locking import UnableToAcquireLock
 from sentry.workflow_engine.migration_helpers.issue_alert_conditions import (
     create_event_unique_user_frequency_condition_with_conditions,
     translate_to_data_condition,
@@ -44,52 +37,6 @@
 SKIPPED_CONDITIONS = [Condition.EVERY_EVENT]
 
 
-class UnableToAcquireLockApiError(SentryAPIException):
-    status_code = status.HTTP_400_BAD_REQUEST
-    code = "unable_to_acquire_lock"
-    message = "Unable to acquire lock for issue alert migration."
-
-
-def ensure_default_error_detector(project: Project) -> Detector:
-    """
-    Ensure that the default error detector exists for a project.
-    If the Detector doesn't already exist, we try to acquire a lock to avoid double-creating,
-    and UnableToAcquireLockApiError if that fails.
-    """
-    # If it already exists, life is simple and we can return immediately.
-    # If there happen to be duplicates, we prefer the oldest.
-    existing = (
-        Detector.objects.filter(type=ErrorGroupType.slug, project=project).order_by("id").first()
-    )
-    if existing:
-        return existing
-
-    # If we may need to create it, we acquire a lock to avoid double-creating.
-    # There isn't a unique constraint on the detector, so we can't rely on get_or_create
-    # to avoid duplicates.
-    # However, by only locking during the one-time creation, the window for a race condition is small.
-    lock = locks.get(
-        f"workflow-engine-project-error-detector:{project.id}",
-        duration=2,
-        name="workflow_engine_default_error_detector",
-    )
-    try:
-        with (
-            # Creation should be fast, so it's worth blocking a little rather
-            # than failing a request.
-            lock.blocking_acquire(initial_delay=0.1, timeout=3),
-            transaction.atomic(router.db_for_write(Detector)),
-        ):
-            detector, _ = Detector.objects.get_or_create(
-                type=ErrorGroupType.slug,
-                project=project,
-                defaults={"config": {}, "name": ERROR_DETECTOR_NAME},
-            )
-            return detector
-    except UnableToAcquireLock:
-        raise UnableToAcquireLockApiError
-
-
 class IssueAlertMigrator:
     def __init__(
         self,
@@ -145,6 +92,7 @@ def _create_detector_lookup(self) -> Detector | None:
                 project=self.project,
                 defaults={"config": {}, "name": ERROR_DETECTOR_NAME},
             )
+            # TODO: get_or_create issue stream detector
             AlertRuleDetector.objects.get_or_create(detector=error_detector, rule_id=self.rule.id)
 
         return error_detector

src/sentry/workflow_engine/processors/detector.py

@@ -6,25 +6,101 @@
 from typing import NamedTuple
 
 import sentry_sdk
+from django.db import router, transaction
+from rest_framework import status
 
 from sentry import options
+from sentry.api.exceptions import SentryAPIException
 from sentry.db.models.manager.base_query_set import BaseQuerySet
 from sentry.grouping.grouptype import ErrorGroupType
+from sentry.issues import grouptype
 from sentry.issues.issue_occurrence import IssueOccurrence
 from sentry.issues.producer import PayloadType, produce_occurrence_to_kafka
+from sentry.locks import locks
 from sentry.models.group import Group
+from sentry.models.project import Project
 from sentry.services.eventstore.models import GroupEvent
 from sentry.utils import metrics
+from sentry.utils.locking import UnableToAcquireLock
 from sentry.workflow_engine.models import DataPacket, Detector
 from sentry.workflow_engine.models.detector_group import DetectorGroup
 from sentry.workflow_engine.types import (
+    ERROR_DETECTOR_NAME,
+    ISSUE_STREAM_DETECTOR_NAME,
     DetectorEvaluationResult,
     DetectorGroupKey,
     WorkflowEventData,
 )
+from sentry.workflow_engine.typings.grouptype import IssueStreamGroupType
 
 logger = logging.getLogger(__name__)
 
+VALID_DEFAULT_DETECTOR_TYPES = [ErrorGroupType.slug, IssueStreamGroupType.slug]
+
+
+class UnableToAcquireLockApiError(SentryAPIException):
+    status_code = status.HTTP_400_BAD_REQUEST
+    code = "unable_to_acquire_lock"
+    message = "Unable to acquire lock for issue alert migration."
+
+
+def _ensure_detector(project: Project, type: str) -> Detector:
+    """
+    Ensure that a detector of a given type exists for a project.
+    If the Detector doesn't already exist, we try to acquire a lock to avoid double-creating,
+    and UnableToAcquireLockApiError if that fails.
+    """
+    group_type = grouptype.registry.get_by_slug(type)
+    if not group_type:
+        raise ValueError(f"Group type {type} not registered")
+    slug = group_type.slug
+    if slug not in VALID_DEFAULT_DETECTOR_TYPES:
+        raise ValueError(f"Invalid default detector type: {slug}")
+
+    # If it already exists, life is simple and we can return immediately.
+    # If there happen to be duplicates, we prefer the oldest.
+    existing = Detector.objects.filter(type=slug, project=project).order_by("id").first()
+    if existing:
+        return existing
+
+    # If we may need to create it, we acquire a lock to avoid double-creating.
+    # There isn't a unique constraint on the detector, so we can't rely on get_or_create
+    # to avoid duplicates.
+    # However, by only locking during the one-time creation, the window for a race condition is small.
+    lock = locks.get(
+        f"workflow-engine-project-{slug}-detector:{project.id}",
+        duration=2,
+        name=f"workflow_engine_default_{slug}_detector",
+    )
+    try:
+        with (
+            # Creation should be fast, so it's worth blocking a little rather
+            # than failing a request.
+            lock.blocking_acquire(initial_delay=0.1, timeout=3),
+            transaction.atomic(router.db_for_write(Detector)),
+        ):
+            detector, _ = Detector.objects.get_or_create(
+                type=slug,
+                project=project,
+                defaults={
+                    "config": {},
+                    "name": (
+                        ERROR_DETECTOR_NAME
+                        if slug == ErrorGroupType.slug
+                        else ISSUE_STREAM_DETECTOR_NAME
+                    ),
+                },
+            )
+            return detector
+    except UnableToAcquireLock:
+        raise UnableToAcquireLockApiError
+
+
+def ensure_default_detectors(project: Project) -> tuple[Detector, Detector]:
+    return _ensure_detector(project, ErrorGroupType.slug), _ensure_detector(
+        project, IssueStreamGroupType.slug
+    )
+
 
 def get_detector_by_event(event_data: WorkflowEventData) -> Detector:
     evt = event_data.event

src/sentry/workflow_engine/types.py

@@ -26,6 +26,7 @@
 T = TypeVar("T")
 
 ERROR_DETECTOR_NAME = "Error Monitor"
+ISSUE_STREAM_DETECTOR_NAME = "Issue Stream"
 
 
 class DetectorException(Exception):

src/sentry/workflow_engine/typings/__init__.py

@@ -0,0 +1,3 @@
+from .grouptype import IssueStreamGroupType
+
+__all__ = ["IssueStreamGroupType"]

src/sentry/workflow_engine/typings/grouptype.py

@@ -0,0 +1,20 @@
+from dataclasses import dataclass
+
+from sentry.issues.grouptype import GroupCategory, GroupType
+
+
+# hidden group type, used for issue stream detector
+@dataclass(frozen=True)
+class IssueStreamGroupType(GroupType):
+    type_id = -1
+    slug = "issue_stream"
+    description = "Issue Stream"
+    category = GroupCategory.ERROR.value
+    category_v2 = GroupCategory.ERROR.value
+    released = False
+    in_default_search = False
+    enable_auto_resolve = False
+    enable_escalation_detection = False
+    enable_status_change_workflow_notifications = False
+    enable_workflow_notifications = False
+    enable_user_priority_changes = False

tests/sentry/projects/project_rules/test_creator.py

@@ -7,9 +7,6 @@
 from sentry.testutils.cases import TestCase
 from sentry.testutils.helpers.features import with_feature
 from sentry.types.actor import Actor
-from sentry.workflow_engine.migration_helpers.issue_alert_migration import (
-    UnableToAcquireLockApiError,
-)
 from sentry.workflow_engine.models import (
     Action,
     AlertRuleDetector,
@@ -18,6 +15,9 @@
     WorkflowDataConditionGroup,
 )
 from sentry.workflow_engine.models.data_condition import Condition
+from sentry.workflow_engine.models.detector import Detector
+from sentry.workflow_engine.processors.detector import UnableToAcquireLockApiError
+from sentry.workflow_engine.typings.grouptype import IssueStreamGroupType
 
 
 class TestProjectRuleCreator(TestCase):
@@ -123,6 +123,8 @@ def test_dual_create_workflow_engine(self) -> None:
         assert detector.project_id == self.project.id
         assert detector.type == ErrorGroupType.slug
 
+        assert Detector.objects.get(project=self.project, type=IssueStreamGroupType.slug)
+
         workflow = alert_rule_workflow.workflow
         assert workflow.config["frequency"] == 5
         assert workflow.owner_user_id == self.user.id

tests/sentry/workflow_engine/migration_helpers/test_issue_alert_migration.py

@@ -22,11 +22,7 @@
 from sentry.testutils.cases import TestCase
 from sentry.testutils.helpers import install_slack
 from sentry.utils.locking import UnableToAcquireLock
-from sentry.workflow_engine.migration_helpers.issue_alert_migration import (
-    IssueAlertMigrator,
-    UnableToAcquireLockApiError,
-    ensure_default_error_detector,
-)
+from sentry.workflow_engine.migration_helpers.issue_alert_migration import IssueAlertMigrator
 from sentry.workflow_engine.models import (
     Action,
     AlertRuleDetector,
@@ -40,6 +36,12 @@
     WorkflowDataConditionGroup,
 )
 from sentry.workflow_engine.models.data_condition import Condition
+from sentry.workflow_engine.processors.detector import (
+    UnableToAcquireLockApiError,
+    ensure_default_detectors,
+)
+from sentry.workflow_engine.types import ERROR_DETECTOR_NAME, ISSUE_STREAM_DETECTOR_NAME
+from sentry.workflow_engine.typings.grouptype import IssueStreamGroupType
 
 
 class IssueAlertMigratorTest(TestCase):
@@ -456,29 +458,35 @@ def test_dry_run__action_validation_fails(self) -> None:
         self.assert_nothing_migrated(self.issue_alert)
 
 
-class TestEnsureDefaultErrorDetector(TestCase):
-    def test_ensure_default_error_detector(self) -> None:
+class TestEnsureDefaultDetectors(TestCase):
+    def setUp(self) -> None:
+        self.slugs = [ErrorGroupType.slug, IssueStreamGroupType.slug]
+        self.names = [ERROR_DETECTOR_NAME, ISSUE_STREAM_DETECTOR_NAME]
+
+    def test_ensure_default_detector(self) -> None:
         project = self.create_project()
-        detector = ensure_default_error_detector(project)
-        assert detector.name == "Error Monitor"
-        assert detector.project_id == project.id
-        assert detector.type == ErrorGroupType.slug
+        error_detector, issue_stream_detector = ensure_default_detectors(project)
+
+        assert error_detector.name == ERROR_DETECTOR_NAME
+        assert error_detector.project_id == project.id
+        assert error_detector.type == ErrorGroupType.slug
+        assert issue_stream_detector.name == ISSUE_STREAM_DETECTOR_NAME
+        assert issue_stream_detector.project_id == project.id
+        assert issue_stream_detector.type == IssueStreamGroupType.slug
 
-    def test_ensure_default_error_detector__already_exists(self) -> None:
+    def test_ensure_default_detector__already_exists(self) -> None:
         project = self.create_project()
-        detector = ensure_default_error_detector(project)
-        with patch(
-            "sentry.workflow_engine.migration_helpers.issue_alert_migration.locks.get"
-        ) as mock_lock:
-            assert ensure_default_error_detector(project).id == detector.id
+        detectors = ensure_default_detectors(project)
+        with patch("sentry.workflow_engine.processors.detector.locks.get") as mock_lock:
+            default_detectors = ensure_default_detectors(project)
+            assert default_detectors[0].id == detectors[0].id
+            assert default_detectors[1].id == detectors[1].id
             # No lock if it already exists.
             mock_lock.assert_not_called()
 
-    def test_ensure_default_error_detector__lock_fails(self) -> None:
+    def test_ensure_default_detector__lock_fails(self) -> None:
         project = self.create_project()
-        with patch(
-            "sentry.workflow_engine.migration_helpers.issue_alert_migration.locks.get"
-        ) as mock_lock:
+        with patch("sentry.workflow_engine.processors.detector.locks.get") as mock_lock:
             mock_lock.return_value.blocking_acquire.side_effect = UnableToAcquireLock
             with pytest.raises(UnableToAcquireLockApiError):
-                ensure_default_error_detector(project)
+                ensure_default_detectors(project)