ref(workflow_engine): Refactor StatefulDetectorHandler + StatefulGroupingDetectorHandler (#92457)

# Description
- Merge the Grouping handler back into the stateful handler, as I was
progressing in refactoring it away, i found myself hardcoding `none` a
lot for the shared helpers. So refactored this to merge them a bit more
together, this will take away the difficulty for implementing a stateful
detector handler, but not reduce all the complexity in the class (seems
like a reasonable tradeoff for now)
- Update the Metric Alert detector to use the new handler.

Author: saponifi3d

src/sentry/incidents/grouptype.py

@@ -7,19 +7,16 @@
 from sentry import features
 from sentry.incidents.metric_alert_detector import MetricAlertsDetectorValidator
 from sentry.incidents.models.alert_rule import AlertRuleDetectionType, ComparisonDeltaChoices
-from sentry.incidents.utils.types import MetricDetectorUpdate, QuerySubscriptionUpdate
+from sentry.incidents.utils.types import QuerySubscriptionUpdate
 from sentry.issues.grouptype import GroupCategory, GroupType
 from sentry.models.organization import Organization
 from sentry.ratelimits.sliding_windows import Quota
 from sentry.types.group import PriorityLevel
-from sentry.workflow_engine.handlers.detector import (
-    DetectorOccurrence,
-    StatefulGroupingDetectorHandler,
-)
+from sentry.workflow_engine.handlers.detector import DetectorOccurrence, StatefulDetectorHandler
 from sentry.workflow_engine.handlers.detector.base import EvidenceData
 from sentry.workflow_engine.models.data_source import DataPacket
 from sentry.workflow_engine.processors.data_condition_group import ProcessedDataConditionGroup
-from sentry.workflow_engine.types import DetectorGroupKey, DetectorPriorityLevel, DetectorSettings
+from sentry.workflow_engine.types import DetectorPriorityLevel, DetectorSettings
 
 COMPARISON_DELTA_CHOICES: list[None | int] = [choice.value for choice in ComparisonDeltaChoices]
 COMPARISON_DELTA_CHOICES.append(None)
@@ -30,7 +27,7 @@ class MetricIssueEvidenceData(EvidenceData):
     alert_id: int
 
 
-class MetricAlertDetectorHandler(StatefulGroupingDetectorHandler[QuerySubscriptionUpdate, int]):
+class MetricAlertDetectorHandler(StatefulDetectorHandler[QuerySubscriptionUpdate, int]):
     def create_occurrence(
         self,
         evaluation_result: ProcessedDataConditionGroup,
@@ -47,22 +44,12 @@ def create_occurrence(
         )
         return occurrence, {}
 
-    @property
-    def counter_names(self) -> list[str]:
-        # Placeholder for now, this should be a list of counters that we want to update as we go above warning / critical
-        return []
-
     def extract_dedupe_value(self, data_packet: DataPacket[QuerySubscriptionUpdate]) -> int:
         return int(data_packet.packet.get("timestamp", datetime.now(UTC)).timestamp())
 
     def extract_value(self, data_packet: DataPacket[QuerySubscriptionUpdate]) -> int:
         return data_packet.packet["values"]["value"]
 
-    def extract_group_values(
-        self, data_packet: DataPacket[MetricDetectorUpdate]
-    ) -> dict[DetectorGroupKey, int]:
-        return {None: data_packet.packet["values"]["value"]}
-
 
 # Example GroupType and detector handler for metric alerts. We don't create these issues yet, but we'll use something
 # like these when we're sending issues as alerts

src/sentry/incidents/subscription_processor.py

@@ -46,7 +46,6 @@
 )
 from sentry.incidents.utils.types import (
     DATA_SOURCE_SNUBA_QUERY_SUBSCRIPTION,
-    MetricDetectorUpdate,
     QuerySubscriptionUpdate,
 )
 from sentry.models.project import Project
@@ -336,13 +335,13 @@ def process_update(self, subscription_update: QuerySubscriptionUpdate) -> None:
 
         if aggregation_value is not None:
             if has_metric_alert_processing:
-                packet = MetricDetectorUpdate(
+                packet = QuerySubscriptionUpdate(
                     entity=subscription_update.get("entity", ""),
                     subscription_id=subscription_update["subscription_id"],
                     values={"value": aggregation_value},
                     timestamp=self.last_update,
                 )
-                data_packet = DataPacket[MetricDetectorUpdate](
+                data_packet = DataPacket[QuerySubscriptionUpdate](
                     source_id=str(self.subscription.id), packet=packet
                 )
                 # temporarily skip processing any anomaly detection alerts

src/sentry/incidents/utils/types.py

@@ -10,13 +10,6 @@ class QuerySubscriptionUpdate(TypedDict):
     timestamp: datetime
 
 
-class MetricDetectorUpdate(TypedDict):
-    entity: str
-    subscription_id: str
-    values: Any
-    timestamp: datetime
-
-
 class AlertRuleActivationConditionType(Enum):
     RELEASE_CREATION = 0
     DEPLOY_CREATION = 1

src/sentry/workflow_engine/handlers/detector/__init__.py

@@ -4,8 +4,8 @@
     "DetectorHandler",
     "DetectorOccurrence",
     "DetectorStateData",
-    "StatefulGroupingDetectorHandler",
+    "StatefulDetectorHandler",
 ]
 
 from .base import DataPacketEvaluationType, DataPacketType, DetectorHandler, DetectorOccurrence
-from .stateful import DetectorStateData, StatefulGroupingDetectorHandler
+from .stateful import DetectorStateData, StatefulDetectorHandler

src/sentry/workflow_engine/handlers/detector/base.py

@@ -22,7 +22,6 @@
 DataPacketType = TypeVar("DataPacketType")
 DataPacketEvaluationType = TypeVar("DataPacketEvaluationType")
 
-# TODO - get more info about how this is used in issue platform
 EventData = dict[str, Any]
 
 
@@ -79,7 +78,6 @@ def to_issue_occurrence(
         )
 
 
-# TODO - DetectorHandler -> AbstractDetectorHandler? (then DetectorHandler is the base implementation)
 class DetectorHandler(abc.ABC, Generic[DataPacketType, DataPacketEvaluationType]):
     def __init__(self, detector: Detector):
         self.detector = detector
@@ -125,19 +123,20 @@ def create_occurrence(
         pass
 
     @abc.abstractmethod
-    def extract_value(self, data_packet: DataPacket[DataPacketType]) -> DataPacketEvaluationType:
+    def extract_value(
+        self, data_packet: DataPacket[DataPacketType]
+    ) -> DataPacketEvaluationType | dict[DetectorGroupKey, DataPacketEvaluationType]:
         """
         Extracts the evaluation value from the data packet to be processed.
 
         This value is used to determine if the data condition group is in a triggered state.
         """
         pass
 
-    # TODO should this be a required method? :thinking:
     @abc.abstractmethod
     def extract_dedupe_value(self, data_packet: DataPacket[DataPacketType]) -> int:
         """
-        Extracts the deduplication value from a passed data packet. This duplication
+        Extracts the de-duplication value from a passed data packet. This duplication
         value is used to determine if we've already processed data to this point or not.
 
         This is normally a timestamp, but could be any sortable value; (e.g. a sequence number, timestamp, etc).