ref(ACI): Move get_comparison_aggregation_value out of subscription processor (#91893)

Redo of #91783 but filtering out
anomaly detection alerts (we haven't migrated them yet) and only
accessing `detector` if we found it. See the latest 2 commits for these
changes.

Fixes https://sentry.sentry.io/issues/6616374058,
https://sentry.sentry.io/issues/6616374079, and
https://sentry.sentry.io/issues/6616374076

Author: ceorourke

src/sentry/incidents/subscription_processor.py

@@ -11,7 +11,6 @@
 from django.db import router, transaction
 from django.utils import timezone
 from sentry_redis_tools.retrying_cluster import RetryingRedisCluster
-from snuba_sdk import Column, Condition, Limit, Op
 
 from sentry import features
 from sentry.constants import ObjectStatus
@@ -42,7 +41,7 @@
 from sentry.incidents.tasks import handle_trigger_action
 from sentry.incidents.utils.metric_issue_poc import create_or_update_metric_issue
 from sentry.incidents.utils.process_update_helpers import (
-    get_aggregation_value_helper,
+    get_comparison_aggregation_value,
     get_crash_rate_alert_metrics_aggregation_value_helper,
 )
 from sentry.incidents.utils.types import (
@@ -51,20 +50,14 @@
     QuerySubscriptionUpdate,
 )
 from sentry.models.project import Project
-from sentry.search.eap.utils import add_start_end_conditions
 from sentry.seer.anomaly_detection.get_anomaly_data import get_anomaly_data_from_seer
 from sentry.seer.anomaly_detection.utils import anomaly_has_confidence, has_anomaly
 from sentry.snuba.dataset import Dataset
-from sentry.snuba.entity_subscription import (
-    ENTITY_TIME_COLUMNS,
-    get_entity_key_from_query_builder,
-    get_entity_subscription_from_snuba_query,
-)
-from sentry.snuba.models import QuerySubscription, SnubaQuery
+from sentry.snuba.models import QuerySubscription
 from sentry.snuba.subscriptions import delete_snuba_subscription
-from sentry.utils import metrics, redis, snuba_rpc
+from sentry.utils import metrics, redis
 from sentry.utils.dates import to_datetime
-from sentry.workflow_engine.models import DataPacket
+from sentry.workflow_engine.models import DataPacket, Detector
 from sentry.workflow_engine.processors.data_packet import process_data_packets
 
 logger = logging.getLogger(__name__)
@@ -216,106 +209,6 @@ def calculate_resolve_threshold(self, trigger: AlertRuleTrigger) -> float:
         threshold: float = trigger.alert_threshold + resolve_add
         return threshold
 
-    def get_comparison_aggregation_value(
-        self, subscription_update: QuerySubscriptionUpdate
-    ) -> float | None:
-        # NOTE (mifu67): we create this helper because we also use it in the new detector processing flow
-        aggregation_value = get_aggregation_value_helper(subscription_update)
-        if self.alert_rule.comparison_delta is None:
-            return aggregation_value
-
-        # For comparison alerts run a query over the comparison period and use it to calculate the
-        # % change.
-        delta = timedelta(seconds=self.alert_rule.comparison_delta)
-        end = subscription_update["timestamp"] - delta
-        snuba_query = self.subscription.snuba_query
-        start = end - timedelta(seconds=snuba_query.time_window)
-
-        entity_subscription = get_entity_subscription_from_snuba_query(
-            snuba_query,
-            self.subscription.project.organization_id,
-        )
-        dataset = Dataset(snuba_query.dataset)
-        query_type = SnubaQuery.Type(snuba_query.type)
-        project_ids = [self.subscription.project_id]
-
-        comparison_aggregate: None | float = None
-        if query_type == SnubaQuery.Type.PERFORMANCE and dataset == Dataset.EventsAnalyticsPlatform:
-            try:
-                rpc_time_series_request = entity_subscription.build_rpc_request(
-                    query=snuba_query.query,
-                    project_ids=project_ids,
-                    environment=snuba_query.environment,
-                    params={
-                        "organization_id": self.subscription.project.organization.id,
-                        "project_id": project_ids,
-                    },
-                    referrer="subscription_processor.comparison_query",
-                )
-
-                rpc_time_series_request = add_start_end_conditions(
-                    rpc_time_series_request, start, end
-                )
-
-                rpc_response = snuba_rpc.timeseries_rpc([rpc_time_series_request])[0]
-                if len(rpc_response.result_timeseries):
-                    comparison_aggregate = rpc_response.result_timeseries[0].data_points[0].data
-
-            except Exception:
-                logger.exception(
-                    "Failed to run RPC comparison query",
-                    extra={
-                        "alert_rule_id": self.alert_rule.id,
-                        "subscription_id": subscription_update.get("subscription_id"),
-                        "organization_id": self.alert_rule.organization_id,
-                    },
-                )
-                return None
-
-        else:
-            try:
-                # TODO: determine whether we need to include the subscription query_extra here
-                query_builder = entity_subscription.build_query_builder(
-                    query=snuba_query.query,
-                    project_ids=project_ids,
-                    environment=snuba_query.environment,
-                    params={
-                        "organization_id": self.subscription.project.organization.id,
-                        "project_id": project_ids,
-                        "start": start,
-                        "end": end,
-                    },
-                )
-                time_col = ENTITY_TIME_COLUMNS[get_entity_key_from_query_builder(query_builder)]
-                query_builder.add_conditions(
-                    [
-                        Condition(Column(time_col), Op.GTE, start),
-                        Condition(Column(time_col), Op.LT, end),
-                    ]
-                )
-                query_builder.limit = Limit(1)
-                results = query_builder.run_query(
-                    referrer="subscription_processor.comparison_query"
-                )
-                comparison_aggregate = list(results["data"][0].values())[0]
-
-            except Exception:
-                logger.exception(
-                    "Failed to run comparison query",
-                    extra={
-                        "alert_rule_id": self.alert_rule.id,
-                        "subscription_id": subscription_update.get("subscription_id"),
-                        "organization_id": self.alert_rule.organization_id,
-                    },
-                )
-                return None
-
-        if not comparison_aggregate:
-            metrics.incr("incidents.alert_rules.skipping_update_comparison_value_invalid")
-            return None
-
-        return (aggregation_value / comparison_aggregate) * 100
-
     def get_crash_rate_alert_metrics_aggregation_value(
         self, subscription_update: QuerySubscriptionUpdate
     ) -> float | None:
@@ -342,13 +235,22 @@ def get_crash_rate_alert_metrics_aggregation_value(
             self.reset_trigger_counts()
         return aggregation_value
 
-    def get_aggregation_value(self, subscription_update: QuerySubscriptionUpdate) -> float | None:
+    def get_aggregation_value(
+        self, subscription_update: QuerySubscriptionUpdate, comparison_delta: int | None = None
+    ) -> float | None:
         if self.subscription.snuba_query.dataset == Dataset.Metrics.value:
             aggregation_value = self.get_crash_rate_alert_metrics_aggregation_value(
                 subscription_update
             )
         else:
-            aggregation_value = self.get_comparison_aggregation_value(subscription_update)
+            aggregation_value = get_comparison_aggregation_value(
+                subscription_update=subscription_update,
+                snuba_query=self.subscription.snuba_query,
+                organization_id=self.subscription.project.organization.id,
+                project_ids=[self.subscription.project_id],
+                comparison_delta=comparison_delta,
+                alert_rule_id=self.alert_rule.id,
+            )
 
         return aggregation_value
 
@@ -366,14 +268,15 @@ def process_update(self, subscription_update: QuerySubscriptionUpdate) -> None:
         if self.subscription.project.status != ObjectStatus.ACTIVE:
             metrics.incr("incidents.alert_rules.ignore_deleted_project")
             return
-        if dataset == "events" and not features.has(
-            "organizations:incidents", self.subscription.project.organization
-        ):
+
+        organization = self.subscription.project.organization
+
+        if dataset == "events" and not features.has("organizations:incidents", organization):
             # They have downgraded since these subscriptions have been created. So we just ignore updates for now.
             metrics.incr("incidents.alert_rules.ignore_update_missing_incidents")
             return
         elif dataset == "transactions" and not features.has(
-            "organizations:performance-view", self.subscription.project.organization
+            "organizations:performance-view", organization
         ):
             # They have downgraded since these subscriptions have been created. So we just ignore updates for now.
             metrics.incr("incidents.alert_rules.ignore_update_missing_incidents_performance")
@@ -390,8 +293,6 @@ def process_update(self, subscription_update: QuerySubscriptionUpdate) -> None:
             metrics.incr("incidents.alert_rules.skipping_already_processed_update")
             return
 
-        aggregation_value = self.get_aggregation_value(subscription_update)
-
         self.last_update = subscription_update["timestamp"]
 
         if (
@@ -408,11 +309,33 @@ def process_update(self, subscription_update: QuerySubscriptionUpdate) -> None:
                 },
             )
 
+        has_metric_alert_processing = features.has(
+            "organizations:workflow-engine-metric-alert-processing", organization
+        )
+        comparison_delta = None
+
+        if (
+            has_metric_alert_processing
+            and not self.alert_rule.detection_type == AlertRuleDetectionType.DYNAMIC
+        ):
+            try:
+                detector = Detector.objects.get(
+                    data_sources__source_id=str(self.subscription.id),
+                    data_sources__type=DATA_SOURCE_SNUBA_QUERY_SUBSCRIPTION,
+                )
+                comparison_delta = detector.config.get("comparison_delta")
+            except Detector.DoesNotExist:
+                logger.exception(
+                    "Detector not found", extra={"subscription_id": self.subscription.id}
+                )
+
+        else:
+            comparison_delta = self.alert_rule.comparison_delta
+
+        aggregation_value = self.get_aggregation_value(subscription_update, comparison_delta)
+
         if aggregation_value is not None:
-            if features.has(
-                "organizations:workflow-engine-metric-alert-processing",
-                self.subscription.project.organization,
-            ):
+            if has_metric_alert_processing:
                 packet = MetricDetectorUpdate(
                     entity=subscription_update.get("entity", ""),
                     subscription_id=subscription_update["subscription_id"],
@@ -438,10 +361,8 @@ def process_update(self, subscription_update: QuerySubscriptionUpdate) -> None:
                     )
 
         has_anomaly_detection = features.has(
-            "organizations:anomaly-detection-alerts", self.subscription.project.organization
-        ) and features.has(
-            "organizations:anomaly-detection-rollout", self.subscription.project.organization
-        )
+            "organizations:anomaly-detection-alerts", organization
+        ) and features.has("organizations:anomaly-detection-rollout", organization)
 
         alert_triggered_tags = {
             "detection_type": self.alert_rule.detection_type,

src/sentry/incidents/tasks.py

@@ -12,14 +12,10 @@
     IncidentStatus,
     IncidentStatusMethod,
 )
-from sentry.incidents.utils.constants import (
-    INCIDENTS_SNUBA_SUBSCRIPTION_TYPE,
-    SUBSCRIPTION_METRICS_LOGGER,
-)
+from sentry.incidents.utils.constants import INCIDENTS_SNUBA_SUBSCRIPTION_TYPE
 from sentry.incidents.utils.types import QuerySubscriptionUpdate
 from sentry.models.project import Project
 from sentry.silo.base import SiloMode
-from sentry.snuba.dataset import Dataset
 from sentry.snuba.models import QuerySubscription
 from sentry.snuba.query_subscriptions.consumer import register_subscriber
 from sentry.tasks.base import instrumented_task
@@ -31,36 +27,6 @@
 logger = logging.getLogger(__name__)
 
 
-@register_subscriber(SUBSCRIPTION_METRICS_LOGGER)
-def handle_subscription_metrics_logger(
-    subscription_update: QuerySubscriptionUpdate, subscription: QuerySubscription
-) -> None:
-    """
-    Logs results from a `QuerySubscription`.
-    """
-    from sentry.incidents.subscription_processor import SubscriptionProcessor
-
-    try:
-        if subscription.snuba_query.dataset == Dataset.Metrics.value:
-            processor = SubscriptionProcessor(subscription)
-            # XXX: Temporary hack so that we can extract these values without raising an exception
-            processor.reset_trigger_counts = lambda *arg, **kwargs: None  # type: ignore[method-assign]
-            aggregation_value = processor.get_aggregation_value(subscription_update)
-
-            logger.info(
-                "handle_subscription_metrics_logger.message",
-                extra={
-                    "subscription_id": subscription.id,
-                    "dataset": subscription.snuba_query.dataset,
-                    "snuba_subscription_id": subscription.subscription_id,
-                    "result": subscription_update,
-                    "aggregation_value": aggregation_value,
-                },
-            )
-    except Exception:
-        logger.exception("Failed to log subscription results")
-
-
 @register_subscriber(INCIDENTS_SNUBA_SUBSCRIPTION_TYPE)
 def handle_snuba_query_update(
     subscription_update: QuerySubscriptionUpdate, subscription: QuerySubscription

src/sentry/incidents/utils/constants.py

@@ -1,3 +1,2 @@
 INCIDENTS_SNUBA_SUBSCRIPTION_TYPE = "incidents"
 INCIDENT_SNAPSHOT_BATCH_SIZE = 50
-SUBSCRIPTION_METRICS_LOGGER = "subscription_metrics_logger"