feat(ACI): Allow processing dynamic data conditions (#93851)

Enable processing dynamic data conditions through workflow engine.

---------

Co-authored-by: Michelle Fu <michelle.fu@sentry.io>

Author: ceorourke

src/sentry/incidents/grouptype.py

@@ -100,7 +100,11 @@ def extract_dedupe_value(self, data_packet: DataPacket[QuerySubscriptionUpdate])
         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"]
+        # this is a bit of a hack - anomaly detection data packets send extra data we need to pass along
+        values = data_packet.packet["values"]
+        if values.get("value") is not None:
+            return values.get("value")
+        return values
 
     def construct_title(
         self,

src/sentry/incidents/handlers/condition/anomaly_detection_handler.py

@@ -1,18 +1,18 @@
 import logging
-from typing import Any
+from datetime import datetime
+from typing import Any, TypedDict
 
 from django.conf import settings
 
 from sentry.net.http import connection_from_url
-from sentry.seer.anomaly_detection.get_anomaly_data import get_anomaly_data_from_seer
 from sentry.seer.anomaly_detection.types import (
     AnomalyDetectionSeasonality,
     AnomalyDetectionSensitivity,
     AnomalyDetectionThresholdType,
     AnomalyType,
 )
 from sentry.snuba.models import QuerySubscription
-from sentry.workflow_engine.models import Condition, DataPacket
+from sentry.workflow_engine.models import Condition
 from sentry.workflow_engine.models.data_condition import DataConditionEvaluationException
 from sentry.workflow_engine.registry import condition_handler_registry
 from sentry.workflow_engine.types import DataConditionHandler, DetectorPriorityLevel
@@ -31,8 +31,15 @@
 }
 
 
+class AnomalyDetectionUpdate(TypedDict):
+    value: int
+    source_id: int
+    subscription_id: int
+    timestamp: datetime
+
+
 @condition_handler_registry.register(Condition.ANOMALY_DETECTION)
-class AnomalyDetectionHandler(DataConditionHandler[DataPacket]):
+class AnomalyDetectionHandler(DataConditionHandler[AnomalyDetectionUpdate]):
     group = DataConditionHandler.Group.DETECTOR_TRIGGER
     comparison_json_schema = {
         "type": "object",
@@ -55,21 +62,24 @@ class AnomalyDetectionHandler(DataConditionHandler[DataPacket]):
     }
 
     @staticmethod
-    def evaluate_value(update: DataPacket, comparison: Any) -> DetectorPriorityLevel:
+    def evaluate_value(update: AnomalyDetectionUpdate, comparison: Any) -> DetectorPriorityLevel:
+        from sentry.seer.anomaly_detection.get_anomaly_data import get_anomaly_data_from_seer
+
         sensitivity = comparison["sensitivity"]
         seasonality = comparison["seasonality"]
         threshold_type = comparison["threshold_type"]
 
-        subscription: QuerySubscription = QuerySubscription.objects.get(id=int(update.source_id))
+        source_id = update.get("source_id")
+        assert source_id
 
-        subscription_update = update.packet
+        subscription: QuerySubscription = QuerySubscription.objects.get(id=int(source_id))
 
         anomaly_data = get_anomaly_data_from_seer(
             sensitivity=sensitivity,
             seasonality=seasonality,
             threshold_type=threshold_type,
             subscription=subscription,
-            subscription_update=subscription_update,
+            subscription_update=update,
         )
         # covers both None and []
         if not anomaly_data:

src/sentry/incidents/subscription_processor.py

@@ -13,6 +13,7 @@
 from sentry_redis_tools.retrying_cluster import RetryingRedisCluster
 
 from sentry import features
+from sentry.api.exceptions import ResourceDoesNotExist
 from sentry.constants import ObjectStatus
 from sentry.incidents.logic import (
     CRITICAL_TRIGGER_LABEL,
@@ -50,6 +51,9 @@
 )
 from sentry.models.project import Project
 from sentry.seer.anomaly_detection.get_anomaly_data import get_anomaly_data_from_seer_legacy
+from sentry.seer.anomaly_detection.get_historical_anomalies import (
+    get_anomaly_evaluation_from_workflow_engine,
+)
 from sentry.seer.anomaly_detection.utils import anomaly_has_confidence, has_anomaly
 from sentry.snuba.dataset import Dataset
 from sentry.snuba.models import QuerySubscription
@@ -253,6 +257,40 @@ def get_aggregation_value(
 
         return aggregation_value
 
+    def handle_trigger_anomalies(
+        self,
+        has_anomaly: bool,
+        trigger: AlertRuleTrigger,
+        aggregation_value: float,
+        fired_incident_triggers: list[IncidentTrigger],
+    ) -> list[IncidentTrigger]:
+        trigger_matches_status = self.check_trigger_matches_status(trigger, TriggerStatus.ACTIVE)
+
+        if has_anomaly and not trigger_matches_status:
+            metrics.incr(
+                "incidents.alert_rules.threshold.alert",
+                tags={"detection_type": self.alert_rule.detection_type},
+            )
+            incident_trigger = self.trigger_alert_threshold(trigger, aggregation_value)
+            if incident_trigger is not None:
+                fired_incident_triggers.append(incident_trigger)
+        else:
+            self.trigger_alert_counts[trigger.id] = 0
+
+        if not has_anomaly and self.active_incident and trigger_matches_status:
+            metrics.incr(
+                "incidents.alert_rules.threshold.resolve",
+                tags={"detection_type": self.alert_rule.detection_type},
+            )
+            incident_trigger = self.trigger_resolve_threshold(trigger, aggregation_value)
+
+            if incident_trigger is not None:
+                fired_incident_triggers.append(incident_trigger)
+        else:
+            self.trigger_resolve_counts[trigger.id] = 0
+
+        return fired_incident_triggers
+
     def process_update(self, subscription_update: QuerySubscriptionUpdate) -> None:
         """
         This is the core processing method utilized when Query Subscription Consumer fetches updates from kafka
@@ -311,12 +349,14 @@ def process_update(self, subscription_update: QuerySubscriptionUpdate) -> None:
         has_metric_alert_processing = features.has(
             "organizations:workflow-engine-metric-alert-processing", organization
         )
+        has_anomaly_detection = features.has(
+            "organizations:anomaly-detection-alerts", organization
+        ) and features.has("organizations:anomaly-detection-rollout", organization)
+
         comparison_delta = None
+        detector = None
 
-        if (
-            has_metric_alert_processing
-            and not self.alert_rule.detection_type == AlertRuleDetectionType.DYNAMIC
-        ):
+        if has_metric_alert_processing:
             try:
                 detector = Detector.objects.get(
                     data_sources__source_id=str(self.subscription.id),
@@ -335,51 +375,51 @@ def process_update(self, subscription_update: QuerySubscriptionUpdate) -> None:
 
         if aggregation_value is not None:
             if has_metric_alert_processing:
-                packet = QuerySubscriptionUpdate(
-                    entity=subscription_update.get("entity", ""),
-                    subscription_id=subscription_update["subscription_id"],
-                    values={"value": aggregation_value},
-                    timestamp=self.last_update,
-                )
+                if self.alert_rule.detection_type == AlertRuleDetectionType.DYNAMIC:
+                    packet = QuerySubscriptionUpdate(
+                        entity=subscription_update.get("entity", ""),
+                        subscription_id=subscription_update["subscription_id"],
+                        values={
+                            "values": {
+                                "value": aggregation_value,
+                                "source_id": str(self.subscription.id),
+                                "subscription_id": subscription_update["subscription_id"],
+                                "timestamp": self.last_update,
+                            },
+                        },
+                        timestamp=self.last_update,
+                    )
+                else:
+                    packet = QuerySubscriptionUpdate(
+                        entity=subscription_update.get("entity", ""),
+                        subscription_id=subscription_update["subscription_id"],
+                        values={"value": aggregation_value},
+                        timestamp=self.last_update,
+                    )
                 data_packet = DataPacket[QuerySubscriptionUpdate](
                     source_id=str(self.subscription.id), packet=packet
                 )
-                # temporarily skip processing any anomaly detection alerts
-                if self.alert_rule.detection_type != AlertRuleDetectionType.DYNAMIC:
-                    results = process_data_packets(
-                        [data_packet], DATA_SOURCE_SNUBA_QUERY_SUBSCRIPTION
+                results = process_data_packets([data_packet], DATA_SOURCE_SNUBA_QUERY_SUBSCRIPTION)
+                if features.has(
+                    "organizations:workflow-engine-metric-alert-dual-processing-logs",
+                    self.alert_rule.organization,
+                ):
+                    logger.info(
+                        "dual processing results for alert rule",
+                        extra={
+                            "results": results,
+                            "num_results": len(results),
+                            "value": aggregation_value,
+                            "rule_id": self.alert_rule.id,
+                        },
                     )
-                    if features.has(
-                        "organizations:workflow-engine-metric-alert-dual-processing-logs",
-                        self.alert_rule.organization,
-                    ):
-                        logger.info(
-                            "dual processing results for alert rule",
-                            extra={
-                                "results": results,
-                                "num_results": len(results),
-                                "value": aggregation_value,
-                                "rule_id": self.alert_rule.id,
-                            },
-                        )
-
-        has_anomaly_detection = features.has(
-            "organizations:anomaly-detection-alerts", organization
-        ) and features.has("organizations:anomaly-detection-rollout", organization)
 
         potential_anomalies = None
         if (
             has_anomaly_detection
             and self.alert_rule.detection_type == AlertRuleDetectionType.DYNAMIC
+            and not has_metric_alert_processing
         ):
-            logger.info(
-                "Raw subscription update",
-                extra={
-                    "result": subscription_update,
-                    "aggregation_value": aggregation_value,
-                    "rule_id": self.alert_rule.id,
-                },
-            )
             with metrics.timer(
                 "incidents.subscription_processor.process_update.get_anomaly_data_from_seer_legacy"
             ):
@@ -390,28 +430,37 @@ def process_update(self, subscription_update: QuerySubscriptionUpdate) -> None:
                     aggregation_value=aggregation_value,
                 )
             if potential_anomalies is None:
-                logger.info(
-                    "No potential anomalies found",
-                    extra={
-                        "subscription_id": self.subscription.id,
-                        "dataset": self.alert_rule.snuba_query.dataset,
-                        "organization_id": self.subscription.project.organization.id,
-                        "project_id": self.subscription.project_id,
-                        "alert_rule_id": self.alert_rule.id,
-                    },
-                )
                 return
 
         if aggregation_value is None:
             metrics.incr("incidents.alert_rules.skipping_update_invalid_aggregation_value")
             return
 
-        fired_incident_triggers = []
+        fired_incident_triggers: list[IncidentTrigger] = []
         with transaction.atomic(router.db_for_write(AlertRule)):
             # Triggers is the threshold - NOT an instance of a trigger
             metrics_incremented = False
             for trigger in self.triggers:
-                if potential_anomalies:
+                # dual processing of anomaly detection alerts
+                if (
+                    has_anomaly_detection
+                    and has_metric_alert_processing
+                    and self.alert_rule.detection_type == AlertRuleDetectionType.DYNAMIC
+                ):
+                    if not detector:
+                        raise ResourceDoesNotExist("Detector not found, cannot evaluate anomaly")
+
+                    is_anomalous = get_anomaly_evaluation_from_workflow_engine(detector, results)
+                    if is_anomalous is None:
+                        # we only care about True and False — None indicates no change
+                        continue
+
+                    assert isinstance(is_anomalous, bool)
+                    fired_incident_triggers = self.handle_trigger_anomalies(
+                        is_anomalous, trigger, aggregation_value, fired_incident_triggers
+                    )
+
+                elif potential_anomalies:
                     # NOTE: There should only be one anomaly in the list
                     for potential_anomaly in potential_anomalies:
                         # check to see if we have enough data for the dynamic alert rule now
@@ -425,38 +474,10 @@ def process_update(self, subscription_update: QuerySubscriptionUpdate) -> None:
                                 # we don't need to check if the alert should fire if the alert can't fire yet
                                 continue
 
-                        if has_anomaly(
-                            potential_anomaly, trigger.label
-                        ) and not self.check_trigger_matches_status(trigger, TriggerStatus.ACTIVE):
-                            metrics.incr(
-                                "incidents.alert_rules.threshold.alert",
-                                tags={"detection_type": self.alert_rule.detection_type},
-                            )
-                            incident_trigger = self.trigger_alert_threshold(
-                                trigger, aggregation_value
-                            )
-                            if incident_trigger is not None:
-                                fired_incident_triggers.append(incident_trigger)
-                        else:
-                            self.trigger_alert_counts[trigger.id] = 0
-
-                        if (
-                            not has_anomaly(potential_anomaly, trigger.label)
-                            and self.active_incident
-                            and self.check_trigger_matches_status(trigger, TriggerStatus.ACTIVE)
-                        ):
-                            metrics.incr(
-                                "incidents.alert_rules.threshold.resolve",
-                                tags={"detection_type": self.alert_rule.detection_type},
-                            )
-                            incident_trigger = self.trigger_resolve_threshold(
-                                trigger, aggregation_value
-                            )
-
-                            if incident_trigger is not None:
-                                fired_incident_triggers.append(incident_trigger)
-                        else:
-                            self.trigger_resolve_counts[trigger.id] = 0
+                        is_anomalous = has_anomaly(potential_anomaly, trigger.label)
+                        fired_incident_triggers = self.handle_trigger_anomalies(
+                            is_anomalous, trigger, aggregation_value, fired_incident_triggers
+                        )
                 else:
                     # OVER/UNDER value trigger
                     alert_operator, resolve_operator = self.THRESHOLD_TYPE_OPERATORS[

src/sentry/seer/anomaly_detection/get_anomaly_data.py

@@ -4,8 +4,8 @@
 from urllib3.exceptions import MaxRetryError, TimeoutError
 
 from sentry.conf.server import SEER_ANOMALY_DETECTION_ENDPOINT_URL
+from sentry.incidents.handlers.condition.anomaly_detection_handler import AnomalyDetectionUpdate
 from sentry.incidents.models.alert_rule import AlertRule
-from sentry.incidents.utils.types import QuerySubscriptionUpdate
 from sentry.net.http import connection_from_url
 from sentry.seer.anomaly_detection.types import (
     AlertInSeer,
@@ -166,10 +166,10 @@ def get_anomaly_data_from_seer(
     seasonality: AnomalyDetectionSeasonality,
     threshold_type: AnomalyDetectionThresholdType,
     subscription: QuerySubscription,
-    subscription_update: QuerySubscriptionUpdate,
+    subscription_update: AnomalyDetectionUpdate,
 ) -> list[TimeSeriesPoint] | None:
     snuba_query: SnubaQuery = subscription.snuba_query
-    aggregation_value = subscription_update["values"].get("value")
+    aggregation_value = subscription_update.get("value")
     source_id = subscription.id
     source_type = DataSourceType.SNUBA_QUERY_SUBSCRIPTION
     if aggregation_value is None:
@@ -185,6 +185,8 @@ def get_anomaly_data_from_seer(
         "source_id": source_id,
         "source_type": source_type,
     }
+    timestamp = subscription_update.get("timestamp")
+    assert timestamp
 
     anomaly_detection_config = AnomalyDetectionConfig(
         time_period=int(snuba_query.time_window / 60),
@@ -195,9 +197,7 @@ def get_anomaly_data_from_seer(
     context = AlertInSeer(
         source_id=source_id,
         source_type=source_type,
-        cur_window=TimeSeriesPoint(
-            timestamp=subscription_update["timestamp"].timestamp(), value=aggregation_value
-        ),
+        cur_window=TimeSeriesPoint(timestamp=timestamp.timestamp(), value=aggregation_value),
     )
     detect_anomalies_request = DetectAnomaliesRequest(
         organization_id=subscription.project.organization.id,