feat(aci milestone 3): anomaly detection condition handler

src/sentry/incidents/grouptype.py

@@ -5,6 +5,7 @@
 from typing import Any
 
 from sentry import features
+from sentry.incidents.handlers.condition import *  # noqa
 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

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

@@ -0,0 +1,5 @@
+__all__ = [
+    "AnomalyDetectionHandler",
+]
+
+from .anomaly_detection_handler import AnomalyDetectionHandler

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

@@ -0,0 +1,89 @@
+import logging
+from typing import Any
+
+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.registry import condition_handler_registry
+from sentry.workflow_engine.types import DataConditionHandler, DetectorPriorityLevel
+
+logger = logging.getLogger(__name__)
+
+SEER_ANOMALY_DETECTION_CONNECTION_POOL = connection_from_url(
+    settings.SEER_ANOMALY_DETECTION_URL,
+    timeout=settings.SEER_ANOMALY_DETECTION_TIMEOUT,
+)
+
+SEER_EVALUATION_TO_DETECTOR_PRIORITY = {
+    AnomalyType.HIGH_CONFIDENCE.value: DetectorPriorityLevel.HIGH,
+    AnomalyType.LOW_CONFIDENCE.value: DetectorPriorityLevel.MEDIUM,
+    AnomalyType.NONE.value: DetectorPriorityLevel.OK,
+}
+
+
+# placeholder until we create this in the workflow engine model
+class DetectorError(Exception):
+    pass
+
+
+@condition_handler_registry.register(Condition.ANOMALY_DETECTION)
+class AnomalyDetectionHandler(DataConditionHandler[DataPacket]):
+    group = DataConditionHandler.Group.DETECTOR_TRIGGER
+    comparison_json_schema = {
+        "type": "object",
+        "properties": {
+            "sensitivity": {
+                "type": "string",
+                "enum": [*AnomalyDetectionSensitivity],
+            },
+            "seasonality": {
+                "type": "string",
+                "enum": [*AnomalyDetectionSeasonality],
+            },
+            "threshold_type": {
+                "type": "integer",
+                "enum": [*AnomalyDetectionThresholdType],
+            },
+        },
+        "required": ["sensitivity", "seasonality", "threshold_type"],
+        "additionalProperties": False,
+    }
+
+    @staticmethod
+    def evaluate_value(update: DataPacket, comparison: Any) -> DetectorPriorityLevel:
+        sensitivity = comparison["sensitivity"]
+        seasonality = comparison["seasonality"]
+        threshold_type = comparison["threshold_type"]
+
+        subscription: QuerySubscription = QuerySubscription.objects.get(id=update.source_id)
+
+        subscription_update = update.packet
+
+        anomaly_data = get_anomaly_data_from_seer(
+            sensitivity=sensitivity,
+            seasonality=seasonality,
+            threshold_type=threshold_type,
+            subscription=subscription,
+            subscription_update=subscription_update,
+        )
+        # covers both None and []
+        if not anomaly_data:
+            # something went wrong during evaluation
+            raise DetectorError("Error during Seer data evaluation process.")
+
+        anomaly_type = anomaly_data[0].get("anomaly", {}).get("anomaly_type")
+        if anomaly_type == AnomalyType.NO_DATA.value:
+            raise DetectorError("Project doesn't have enough data for detector to evaluate")
+        elif anomaly_type is None:
+            raise DetectorError("Seer response contained no evaluation data")
+
+        return SEER_EVALUATION_TO_DETECTOR_PRIORITY[anomaly_type]

src/sentry/incidents/subscription_processor.py

@@ -50,7 +50,7 @@
     QuerySubscriptionUpdate,
 )
 from sentry.models.project import Project
-from sentry.seer.anomaly_detection.get_anomaly_data import get_anomaly_data_from_seer
+from sentry.seer.anomaly_detection.get_anomaly_data import get_anomaly_data_from_seer_legacy
 from sentry.seer.anomaly_detection.utils import anomaly_has_confidence, has_anomaly
 from sentry.snuba.dataset import Dataset
 from sentry.snuba.models import QuerySubscription
@@ -382,9 +382,9 @@ def process_update(self, subscription_update: QuerySubscriptionUpdate) -> None:
                 },
             )
             with metrics.timer(
-                "incidents.subscription_processor.process_update.get_anomaly_data_from_seer"
+                "incidents.subscription_processor.process_update.get_anomaly_data_from_seer_legacy"
             ):
-                potential_anomalies = get_anomaly_data_from_seer(
+                potential_anomalies = get_anomaly_data_from_seer_legacy(
                     alert_rule=self.alert_rule,
                     subscription=self.subscription,
                     last_update=self.last_update.timestamp(),

src/sentry/migrations/0909_increase_email_model_email_field_length.py

@@ -0,0 +1,39 @@
+# Generated by Django 5.2.1 on 2025-05-21 23:05
+
+from django.db import migrations, models
+
+import sentry.db.models.fields.citext
+from sentry.new_migrations.migrations import CheckedMigration
+
+
+class Migration(CheckedMigration):
+    # This flag is used to mark that a migration shouldn't be automatically run in production.
+    # This should only be used for operations where it's safe to run the migration after your
+    # code has deployed. So this should not be used for most operations that alter the schema
+    # of a table.
+    # Here are some things that make sense to mark as post deployment:
+    # - Large data migrations. Typically we want these to be run manually so that they can be
+    #   monitored and not block the deploy for a long period of time while they run.
+    # - Adding indexes to large tables. Since this can take a long time, we'd generally prefer to
+    #   run this outside deployments so that we don't block them. Note that while adding an index
+    #   is a schema change, it's completely safe to run the operation after the code has deployed.
+    # Once deployed, run these manually via: https://develop.sentry.dev/database-migrations/#migration-deployment
+
+    is_post_deployment = False
+
+    dependencies = [
+        ("sentry", "0908_increase_email_field_length"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="email",
+            name="email",
+            field=sentry.db.models.fields.citext.CIEmailField(max_length=200, unique=True),
+        ),
+        migrations.AlterField(
+            model_name="useremail",
+            name="email",
+            field=models.EmailField(max_length=200),
+        ),
+    ]

src/sentry/seer/anomaly_detection/get_anomaly_data.py

@@ -5,17 +5,22 @@
 
 from sentry.conf.server import SEER_ANOMALY_DETECTION_ENDPOINT_URL
 from sentry.incidents.models.alert_rule import AlertRule
+from sentry.incidents.utils.types import MetricDetectorUpdate
 from sentry.net.http import connection_from_url
 from sentry.seer.anomaly_detection.types import (
     AlertInSeer,
     AnomalyDetectionConfig,
+    AnomalyDetectionSeasonality,
+    AnomalyDetectionSensitivity,
+    AnomalyDetectionThresholdType,
+    DataSourceType,
     DetectAnomaliesRequest,
     DetectAnomaliesResponse,
     TimeSeriesPoint,
 )
 from sentry.seer.anomaly_detection.utils import translate_direction
 from sentry.seer.signed_seer_api import make_signed_seer_api_request
-from sentry.snuba.models import QuerySubscription
+from sentry.snuba.models import QuerySubscription, SnubaQuery
 from sentry.utils import json, metrics
 from sentry.utils.json import JSONDecodeError
 
@@ -27,7 +32,8 @@
 )
 
 
-def get_anomaly_data_from_seer(
+# TODO: delete this once we deprecate the AlertRule model
+def get_anomaly_data_from_seer_legacy(
     alert_rule: AlertRule,
     subscription: QuerySubscription,
     last_update: float,
@@ -153,3 +159,120 @@ def get_anomaly_data_from_seer(
         )
         return None
     return ts
+
+
+def get_anomaly_data_from_seer(
+    sensitivity: AnomalyDetectionSensitivity,
+    seasonality: AnomalyDetectionSeasonality,
+    threshold_type: AnomalyDetectionThresholdType,
+    subscription: QuerySubscription,
+    subscription_update: MetricDetectorUpdate,
+) -> list[TimeSeriesPoint] | None:
+    snuba_query: SnubaQuery = subscription.snuba_query
+    aggregation_value = subscription_update["values"].get("value")
+    source_id = subscription.id
+    source_type = DataSourceType.SNUBA_QUERY_SUBSCRIPTION
+    if aggregation_value is None:
+        logger.error(
+            "Invalid aggregation value", extra={"source_id": source_id, "source_type": source_type}
+        )
+        return None
+
+    extra_data = {
+        "subscription_id": subscription.id,
+        "organization_id": subscription.project.organization.id,
+        "project_id": subscription.project_id,
+        "source_id": source_id,
+        "source_type": source_type,
+    }
+
+    anomaly_detection_config = AnomalyDetectionConfig(
+        time_period=int(snuba_query.time_window / 60),
+        sensitivity=sensitivity,
+        direction=translate_direction(threshold_type),
+        expected_seasonality=seasonality,
+    )
+    context = AlertInSeer(
+        source_id=source_id,
+        source_type=source_type,
+        cur_window=TimeSeriesPoint(
+            timestamp=subscription_update["timestamp"].timestamp(), value=aggregation_value
+        ),
+    )
+    detect_anomalies_request = DetectAnomaliesRequest(
+        organization_id=subscription.project.organization.id,
+        project_id=subscription.project_id,
+        config=anomaly_detection_config,
+        context=context,
+    )
+    extra_data["dataset"] = snuba_query.dataset
+    try:
+        logger.info("Sending subscription update data to Seer", extra=extra_data)
+        response = make_signed_seer_api_request(
+            SEER_ANOMALY_DETECTION_CONNECTION_POOL,
+            SEER_ANOMALY_DETECTION_ENDPOINT_URL,
+            json.dumps(detect_anomalies_request).encode("utf-8"),
+        )
+    except (TimeoutError, MaxRetryError):
+        logger.warning("Timeout error when hitting anomaly detection endpoint", extra=extra_data)
+        return None
+
+    if response.status > 400:
+        logger.error(
+            "Error when hitting Seer detect anomalies endpoint",
+            extra={
+                "response_data": response.data,
+                **extra_data,
+            },
+        )
+        return None
+    try:
+        decoded_data = response.data.decode("utf-8")
+    except AttributeError:
+        logger.exception(
+            "Failed to parse Seer anomaly detection response",
+            extra={
+                "ad_config": anomaly_detection_config,
+                "context": context,
+                "response_data": response.data,
+                "response_code": response.status,
+            },
+        )
+        return None
+
+    try:
+        results: DetectAnomaliesResponse = json.loads(decoded_data)
+    except JSONDecodeError:
+        logger.exception(
+            "Failed to parse Seer anomaly detection response",
+            extra={
+                "ad_config": anomaly_detection_config,
+                "context": context,
+                "response_data": decoded_data,
+                "response_code": response.status,
+            },
+        )
+        return None
+
+    if not results.get("success"):
+        logger.error(
+            "Error when hitting Seer detect anomalies endpoint",
+            extra={
+                "error_message": results.get("message", ""),
+                **extra_data,
+            },
+        )
+        return None
+
+    ts = results.get("timeseries")
+    if not ts:
+        logger.warning(
+            "Seer anomaly detection response returned no potential anomalies",
+            extra={
+                "ad_config": anomaly_detection_config,
+                "context": context,
+                "response_data": results.get("message"),
+            },
+        )
+        return None
+    return ts

src/sentry/seer/anomaly_detection/types.py

@@ -1,4 +1,4 @@
-from enum import Enum
+from enum import IntEnum, StrEnum
 from typing import NotRequired, TypedDict
 
 
@@ -13,8 +13,16 @@ class TimeSeriesPoint(TypedDict):
     anomaly: NotRequired[Anomaly]
 
 
+class DataSourceType(IntEnum):
+    SNUBA_QUERY_SUBSCRIPTION = 0
+
+
 class AlertInSeer(TypedDict):
-    id: int
+    id: NotRequired[int]
+    source_id: NotRequired[
+        int
+    ]  # during our dual processing rollout, some requests will be sending ID and some will send source_id/source_type
+    source_type: NotRequired[DataSourceType]
     cur_window: NotRequired[TimeSeriesPoint]
 
 
@@ -69,8 +77,35 @@ class DetectAnomaliesResponse(TypedDict):
     timeseries: list[TimeSeriesPoint]
 
 
-class AnomalyType(Enum):
+class AnomalyType(StrEnum):
     HIGH_CONFIDENCE = "anomaly_higher_confidence"
     LOW_CONFIDENCE = "anomaly_lower_confidence"
     NONE = "none"
     NO_DATA = "no_data"
+
+
+class AnomalyDetectionSensitivity(StrEnum):
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+
+
+class AnomalyDetectionSeasonality(StrEnum):
+    """All combinations of multi select fields for anomaly detection alerts
+    We do not anticipate adding more
+    """
+
+    AUTO = "auto"
+    HOURLY = "hourly"
+    DAILY = "daily"
+    WEEKLY = "weekly"
+    HOURLY_DAILY = "hourly_daily"
+    HOURLY_WEEKLY = "hourly_weekly"
+    HOURLY_DAILY_WEEKLY = "hourly_daily_weekly"
+    DAILY_WEEKLY = "daily_weekly"
+
+
+class AnomalyDetectionThresholdType(IntEnum):
+    ABOVE = 0
+    BELOW = 1
+    ABOVE_AND_BELOW = 2