fix(aci): Record DetectorGroup associations for new error groups (#100684)

Guarded behind a flag for now.
We expect this to be 140k-190k writes per hour, expected to double the
scale of DetectorGroup in under a week.

Author: kcons

src/sentry/event_manager.py

@@ -143,6 +143,7 @@
 from sentry.utils.safe import get_path, safe_execute, setdefault_path, trim
 from sentry.utils.sdk import set_span_attribute
 from sentry.utils.tag_normalization import normalized_sdk_tag_from_event
+from sentry.workflow_engine.processors.detector import associate_new_group_with_detector
 
 from .utils.event_tracker import TransactionStageStatus, track_sampled_event
 
@@ -1494,6 +1495,7 @@ def create_group_with_grouphashes(job: Job, grouphashes: list[GroupHash]) -> Gro
             record_new_group_metrics(event)
 
             group = _create_group(project, event, **_get_group_processing_kwargs(job))
+            associate_new_group_with_detector(group)
             add_group_id_to_grouphashes(group, grouphashes)
 
             return GroupInfo(group=group, is_new=True, is_regression=False)

src/sentry/issues/ingest.py

@@ -35,7 +35,8 @@
 from sentry.utils import json, metrics, redis
 from sentry.utils.strings import truncatechars
 from sentry.utils.tag_normalization import normalized_sdk_tag_from_event
-from sentry.workflow_engine.models import DetectorGroup, IncidentGroupOpenPeriod
+from sentry.workflow_engine.models import IncidentGroupOpenPeriod
+from sentry.workflow_engine.processors.detector import associate_new_group_with_detector
 
 issue_rate_limiter = RedisSlidingWindowRateLimiter(
     **settings.SENTRY_ISSUE_PLATFORM_RATE_LIMITER_OPTIONS
@@ -255,10 +256,7 @@ def save_issue_from_occurrence(
                 project, event, primary_hash, **issue_kwargs
             )
             if is_new and occurrence.evidence_data and "detector_id" in occurrence.evidence_data:
-                DetectorGroup.objects.get_or_create(
-                    detector_id=occurrence.evidence_data["detector_id"],
-                    group_id=group.id,
-                )
+                associate_new_group_with_detector(group, occurrence.evidence_data["detector_id"])
 
             open_period = get_latest_open_period(group)
             if open_period is not None:

src/sentry/options/defaults.py

@@ -3097,6 +3097,13 @@
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
 
+register(
+    "workflow_engine.associate_error_detectors",
+    type=Bool,
+    default=False,
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)
+
 register(
     "grouping.grouphash_metadata.ingestion_writes_enabled",
     type=Bool,

src/sentry/workflow_engine/processors/detector.py

@@ -7,12 +7,16 @@
 
 import sentry_sdk
 
+from sentry import options
 from sentry.db.models.manager.base_query_set import BaseQuerySet
+from sentry.grouping.grouptype import ErrorGroupType
 from sentry.issues.issue_occurrence import IssueOccurrence
 from sentry.issues.producer import PayloadType, produce_occurrence_to_kafka
+from sentry.models.group import Group
 from sentry.services.eventstore.models import GroupEvent
 from sentry.utils import metrics
 from sentry.workflow_engine.models import DataPacket, Detector
+from sentry.workflow_engine.models.detector_group import DetectorGroup
 from sentry.workflow_engine.types import (
     DetectorEvaluationResult,
     DetectorGroupKey,
@@ -68,8 +72,6 @@ class _SplitEvents(NamedTuple):
 def _split_events_by_occurrence(
     event_list: list[GroupEvent],
 ) -> _SplitEvents:
-    from sentry.grouping.grouptype import ErrorGroupType
-
     events_with_occurrences: list[tuple[GroupEvent, int]] = []
     error_events: list[GroupEvent] = []  # only error events don't have occurrences
     events_missing_detectors: list[GroupEvent] = []
@@ -116,8 +118,6 @@ def get_detectors_by_groupevents_bulk(
     """
     Given a list of GroupEvents, return a mapping of event_id to Detector.
     """
-    from sentry.grouping.grouptype import ErrorGroupType
-
     if not event_list:
         return {}
 
@@ -281,3 +281,25 @@ def process_detectors[T](
             results.append((detector, detector_results))
 
     return results
+
+
+def associate_new_group_with_detector(group: Group, detector_id: int | None = None) -> bool:
+    """
+    Associate a new Group with it's Detector in the database.
+    If the Group is an error, it can be associated without a detector ID.
+
+    Return whether the group was associated.
+    """
+    if detector_id is None:
+        # For error Groups, we know there is a Detector and we can find it by project.
+        if group.type == ErrorGroupType.type_id:
+            if not options.get("workflow_engine.associate_error_detectors", False):
+                return False
+            detector_id = Detector.get_error_detector_for_project(group.project.id).id
+        else:
+            return False
+    DetectorGroup.objects.get_or_create(
+        detector_id=detector_id,
+        group_id=group.id,
+    )
+    return True

tests/sentry/workflow_engine/processors/test_detector.py

@@ -6,8 +6,9 @@
 import pytest
 from django.utils import timezone
 
+from sentry.grouping.grouptype import ErrorGroupType
 from sentry.incidents.grouptype import MetricIssue
-from sentry.issues.grouptype import PerformanceNPlusOneAPICallsGroupType
+from sentry.issues.grouptype import FeedbackGroup, PerformanceNPlusOneAPICallsGroupType
 from sentry.issues.issue_occurrence import IssueOccurrence
 from sentry.issues.producer import PayloadType
 from sentry.issues.status_change_message import StatusChangeMessage
@@ -23,7 +24,9 @@
 from sentry.workflow_engine.handlers.detector import DetectorStateData
 from sentry.workflow_engine.handlers.detector.stateful import get_redis_client
 from sentry.workflow_engine.models import DataPacket, Detector, DetectorState
+from sentry.workflow_engine.models.detector_group import DetectorGroup
 from sentry.workflow_engine.processors.detector import (
+    associate_new_group_with_detector,
     get_detector_by_event,
     get_detectors_by_groupevents_bulk,
     process_detectors,
@@ -1017,3 +1020,40 @@ def test_mixed_occurrences_missing_detectors(self) -> None:
 
             assert result == {}
             mock_metrics.incr.assert_called_with("workflow_engine.detectors.error", amount=1)
+
+
+class TestAssociateNewGroupWithDetector(TestCase):
+    def setUp(self) -> None:
+        super().setUp()
+        self.metric_detector = self.create_detector(project=self.project, type="metric_issue")
+        self.error_detector = self.create_detector(project=self.project, type="error")
+
+    def test_metrics_group_with_known_detector(self) -> None:
+        group = self.create_group(project=self.project, type=MetricIssue.type_id)
+
+        # Should return True and create DetectorGroup
+        assert associate_new_group_with_detector(group, self.metric_detector.id)
+        assert DetectorGroup.objects.filter(
+            detector_id=self.metric_detector.id, group_id=group.id
+        ).exists()
+
+    def test_error_group_with_feature_disabled(self) -> None:
+        group = self.create_group(project=self.project, type=ErrorGroupType.type_id)
+
+        with self.options({"workflow_engine.associate_error_detectors": False}):
+            assert not associate_new_group_with_detector(group)
+            assert not DetectorGroup.objects.filter(group_id=group.id).exists()
+
+    def test_error_group_with_feature_enabled(self) -> None:
+        group = self.create_group(project=self.project, type=ErrorGroupType.type_id)
+
+        with self.options({"workflow_engine.associate_error_detectors": True}):
+            assert associate_new_group_with_detector(group)
+            assert DetectorGroup.objects.filter(
+                detector_id=self.error_detector.id, group_id=group.id
+            ).exists()
+
+    def test_feedback_group_returns_false(self) -> None:
+        group = self.create_group(project=self.project, type=FeedbackGroup.type_id)
+        assert not associate_new_group_with_detector(group)
+        assert not DetectorGroup.objects.filter(group_id=group.id).exists()