Revert "ref(ACI): Move get_comparison_aggregation_value out of subscription processor (#91783)"

This reverts commit fbc6095.

Co-authored-by: ceorourke <29959063+ceorourke@users.noreply.github.com>

Author: 2 people

src/sentry/incidents/subscription_processor.py

@@ -11,6 +11,7 @@
 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
@@ -41,7 +42,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_comparison_aggregation_value,
+    get_aggregation_value_helper,
     get_crash_rate_alert_metrics_aggregation_value_helper,
 )
 from sentry.incidents.utils.types import (
@@ -50,14 +51,20 @@
     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.models import QuerySubscription
+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.subscriptions import delete_snuba_subscription
-from sentry.utils import metrics, redis
+from sentry.utils import metrics, redis, snuba_rpc
 from sentry.utils.dates import to_datetime
-from sentry.workflow_engine.models import DataPacket, Detector
+from sentry.workflow_engine.models import DataPacket
 from sentry.workflow_engine.processors.data_packet import process_data_packets
 
 logger = logging.getLogger(__name__)
@@ -209,6 +216,106 @@ 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:
@@ -235,22 +342,13 @@ def get_crash_rate_alert_metrics_aggregation_value(
             self.reset_trigger_counts()
         return aggregation_value
 
-    def get_aggregation_value(
-        self, subscription_update: QuerySubscriptionUpdate, comparison_delta: int | None = None
-    ) -> float | None:
+    def get_aggregation_value(self, subscription_update: QuerySubscriptionUpdate) -> 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 = 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,
-            )
+            aggregation_value = self.get_comparison_aggregation_value(subscription_update)
 
         return aggregation_value
 
@@ -268,15 +366,14 @@ def process_update(self, subscription_update: QuerySubscriptionUpdate) -> None:
         if self.subscription.project.status != ObjectStatus.ACTIVE:
             metrics.incr("incidents.alert_rules.ignore_deleted_project")
             return
-
-        organization = self.subscription.project.organization
-
-        if dataset == "events" and not features.has("organizations:incidents", organization):
+        if dataset == "events" and not features.has(
+            "organizations:incidents", self.subscription.project.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", organization
+            "organizations:performance-view", self.subscription.project.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")
@@ -293,6 +390,8 @@ 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 (
@@ -309,30 +408,11 @@ 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:
-            try:
-                detector = Detector.objects.get(
-                    data_sources__source_id=str(self.subscription.id),
-                    data_sources__type=DATA_SOURCE_SNUBA_QUERY_SUBSCRIPTION,
-                )
-            except Detector.DoesNotExist:
-                logger.exception(
-                    "Detector not found", extra={"subscription_id": self.subscription.id}
-                )
-
-            comparison_delta = detector.config.get("comparison_delta")
-        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 has_metric_alert_processing:
+            if features.has(
+                "organizations:workflow-engine-metric-alert-processing",
+                self.subscription.project.organization,
+            ):
                 packet = MetricDetectorUpdate(
                     entity=subscription_update.get("entity", ""),
                     subscription_id=subscription_update["subscription_id"],
@@ -358,8 +438,10 @@ def process_update(self, subscription_update: QuerySubscriptionUpdate) -> None:
                     )
 
         has_anomaly_detection = features.has(
-            "organizations:anomaly-detection-alerts", organization
-        ) and features.has("organizations:anomaly-detection-rollout", organization)
+            "organizations:anomaly-detection-alerts", self.subscription.project.organization
+        ) and features.has(
+            "organizations:anomaly-detection-rollout", self.subscription.project.organization
+        )
 
         potential_anomalies = None
         if (

src/sentry/incidents/tasks.py

@@ -12,10 +12,14 @@
     IncidentStatus,
     IncidentStatusMethod,
 )
-from sentry.incidents.utils.constants import INCIDENTS_SNUBA_SUBSCRIPTION_TYPE
+from sentry.incidents.utils.constants import (
+    INCIDENTS_SNUBA_SUBSCRIPTION_TYPE,
+    SUBSCRIPTION_METRICS_LOGGER,
+)
 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
@@ -27,6 +31,36 @@
 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,2 +1,3 @@
 INCIDENTS_SNUBA_SUBSCRIPTION_TYPE = "incidents"
 INCIDENT_SNAPSHOT_BATCH_SIZE = 50
+SUBSCRIPTION_METRICS_LOGGER = "subscription_metrics_logger"