feat(cra-metrics): Adds crash rate alerts over metrics [INGEST-629] [INGEST-632] (#30400)

* feat(cra-metrics): Adds crash rate alerts over metrics

Adds ability to create crash rate alerts over metrics
for sessions. Adds `EntitySubscription` class that
is an abstraction layer for all entity subscriptions

Author: ahmedetefy

mypy.ini

@@ -59,6 +59,7 @@ files = src/sentry/api/bases/external_actor.py,
         src/sentry/shared_integrations/constants.py,
         src/sentry/snuba/outcomes.py,
         src/sentry/snuba/query_subscription_consumer.py,
+        src/sentry/snuba/entity_subscription.py,
         src/sentry/spans/,
         src/sentry/tasks/app_store_connect.py,
         src/sentry/tasks/low_priority_symbolication.py,

src/sentry/api/serializers/models/incident.py

@@ -11,8 +11,8 @@
     IncidentSeen,
     IncidentSubscription,
 )
+from sentry.snuba.entity_subscription import apply_dataset_query_conditions
 from sentry.snuba.models import QueryDatasets
-from sentry.snuba.tasks import apply_dataset_query_conditions
 
 
 @register(Incident)

src/sentry/exceptions.py

@@ -69,3 +69,11 @@ class InvalidSearchQuery(Exception):
 
 class UnableToAcceptMemberInvitationException(Exception):
     pass
+
+
+class UnsupportedQuerySubscription(Exception):
+    pass
+
+
+class InvalidQuerySubscription(Exception):
+    pass

src/sentry/incidents/endpoints/serializers.py

@@ -1,18 +1,20 @@
 import logging
 import operator
+from copy import copy
 from datetime import timedelta
 
 from django.db import transaction
 from django.utils import timezone
 from django.utils.encoding import force_text
 from rest_framework import serializers
+from snuba_sdk.legacy import json_to_snql
 
 from sentry import analytics
 from sentry.api.fields.actor import ActorField
 from sentry.api.serializers.rest_framework.base import CamelSnakeModelSerializer
 from sentry.api.serializers.rest_framework.environment import EnvironmentField
 from sentry.api.serializers.rest_framework.project import ProjectField
-from sentry.exceptions import InvalidSearchQuery
+from sentry.exceptions import InvalidSearchQuery, UnsupportedQuerySubscription
 from sentry.incidents.logic import (
     CRITICAL_TRIGGER_LABEL,
     WARNING_TRIGGER_LABEL,
@@ -43,10 +45,12 @@
 from sentry.models import OrganizationMember, SentryAppInstallation, Team, User
 from sentry.shared_integrations.exceptions import ApiRateLimitedError
 from sentry.snuba.dataset import Dataset
+from sentry.snuba.entity_subscription import map_aggregate_to_entity_subscription
 from sentry.snuba.models import QueryDatasets, SnubaQueryEventType
 from sentry.snuba.tasks import build_snuba_filter
+from sentry.utils import json
 from sentry.utils.compat import zip
-from sentry.utils.snuba import raw_query
+from sentry.utils.snuba import raw_snql_query
 
 logger = logging.getLogger(__name__)
 
@@ -443,7 +447,8 @@ def validate_dataset(self, dataset):
             )
 
     def validate_event_types(self, event_types):
-        if self.initial_data.get("dataset") == Dataset.Sessions.value:
+        dataset = self.initial_data.get("dataset")
+        if dataset not in [Dataset.Events.value, Dataset.Transactions.value]:
             return []
         try:
             return [SnubaQueryEventType.EventType[event_type.upper()] for event_type in event_types]
@@ -477,13 +482,24 @@ def validate(self, data):
             # the query. We don't use the returned data anywhere, so it doesn't
             # matter which.
             project_id = list(self.context["organization"].project_set.all()[:1])
+
+        try:
+            entity_subscription = map_aggregate_to_entity_subscription(
+                dataset=QueryDatasets(data["dataset"]),
+                aggregate=data["aggregate"],
+                extra_fields={
+                    "org_id": project_id[0].organization_id,
+                    "event_types": data.get("event_types"),
+                },
+            )
+        except UnsupportedQuerySubscription as e:
+            raise serializers.ValidationError(f"{e}")
+
         try:
             snuba_filter = build_snuba_filter(
-                data["dataset"],
+                entity_subscription,
                 data["query"],
-                data["aggregate"],
                 data.get("environment"),
-                data.get("event_types"),
                 params={
                     "project_id": [p.id for p in project_id],
                     "start": timezone.now() - timedelta(minutes=10),
@@ -503,18 +519,35 @@ def validate(self, data):
             dataset = Dataset(data["dataset"].value)
             self._validate_time_window(dataset, data.get("time_window"))
 
+            conditions = copy(snuba_filter.conditions)
+            time_col = entity_subscription.time_col
+            conditions += [
+                [time_col, ">=", snuba_filter.start],
+                [time_col, "<", snuba_filter.end],
+            ]
+
+            body = {
+                "project": project_id[0].id,
+                "project_id": project_id[0].id,
+                "aggregations": snuba_filter.aggregations,
+                "conditions": conditions,
+                "filter_keys": snuba_filter.filter_keys,
+                "having": snuba_filter.having,
+                "dataset": dataset.value,
+                "limit": 1,
+                **entity_subscription.get_entity_extra_params(),
+            }
+
             try:
-                raw_query(
-                    aggregations=snuba_filter.aggregations,
-                    start=snuba_filter.start,
-                    end=snuba_filter.end,
-                    conditions=snuba_filter.conditions,
-                    filter_keys=snuba_filter.filter_keys,
-                    having=snuba_filter.having,
-                    dataset=dataset,
-                    limit=1,
-                    referrer="alertruleserializer.test_query",
+                snql_query = json_to_snql(body, entity_subscription.entity_key.value)
+                snql_query.validate()
+            except Exception as e:
+                raise serializers.ValidationError(
+                    str(e), params={"params": json.dumps(body), "dataset": data["dataset"].value}
                 )
+
+            try:
+                raw_snql_query(snql_query, referrer="alertruleserializer.test_query")
             except Exception:
                 logger.exception("Error while validating snuba alert rule query")
                 raise serializers.ValidationError(
@@ -582,7 +615,7 @@ def _translate_thresholds(self, threshold_type, comparison_delta, triggers, data
 
     @staticmethod
     def _validate_time_window(dataset, time_window):
-        if dataset == Dataset.Sessions:
+        if dataset in [Dataset.Sessions, Dataset.Metrics]:
             # Validate time window
             if time_window not in CRASH_RATE_ALERTS_ALLOWED_TIME_WINDOWS:
                 raise serializers.ValidationError(

src/sentry/incidents/logic.py

@@ -35,6 +35,7 @@
 from sentry.search.events.filter import get_filter
 from sentry.shared_integrations.exceptions import DuplicateDisplayNameError
 from sentry.snuba.dataset import Dataset
+from sentry.snuba.entity_subscription import map_aggregate_to_entity_subscription
 from sentry.snuba.models import QueryDatasets
 from sentry.snuba.subscriptions import (
     bulk_create_snuba_subscriptions,
@@ -288,12 +289,15 @@ def build_incident_query_params(incident, start=None, end=None, windowed_stats=F
         params["project_id"] = project_ids
 
     snuba_query = incident.alert_rule.snuba_query
+    entity_subscription = map_aggregate_to_entity_subscription(
+        dataset=QueryDatasets(snuba_query.dataset),
+        aggregate=snuba_query.aggregate,
+        extra_fields={"org_id": incident.organization.id, "event_types": snuba_query.event_types},
+    )
     snuba_filter = build_snuba_filter(
-        QueryDatasets(snuba_query.dataset),
+        entity_subscription,
         snuba_query.query,
-        snuba_query.aggregate,
         snuba_query.environment,
-        snuba_query.event_types,
         params=params,
     )
 

src/sentry/incidents/subscription_processor.py

@@ -32,7 +32,7 @@
 from sentry.release_health.metrics import reverse_tag_value, tag_key
 from sentry.snuba.dataset import Dataset
 from sentry.snuba.models import QueryDatasets
-from sentry.snuba.tasks import build_snuba_filter
+from sentry.snuba.tasks import build_snuba_filter, map_aggregate_to_entity_subscription
 from sentry.utils import metrics, redis
 from sentry.utils.compat import zip
 from sentry.utils.dates import to_datetime, to_timestamp
@@ -176,13 +176,19 @@ def get_comparison_aggregation_value(self, subscription_update, aggregation_valu
         snuba_query = self.subscription.snuba_query
         start = end - timedelta(seconds=snuba_query.time_window)
 
+        entity_subscription = map_aggregate_to_entity_subscription(
+            dataset=QueryDatasets(snuba_query.dataset),
+            aggregate=snuba_query.aggregate,
+            extra_fields={
+                "org_id": self.subscription.project.organization,
+                "event_types": snuba_query.event_types,
+            },
+        )
         try:
             snuba_filter = build_snuba_filter(
-                QueryDatasets(snuba_query.dataset),
+                entity_subscription,
                 snuba_query.query,
-                snuba_query.aggregate,
                 snuba_query.environment,
-                snuba_query.event_types,
                 params={
                     "project_id": [self.subscription.project_id],
                     "start": start,

src/sentry/snuba/dataset.py

@@ -14,6 +14,9 @@ class Dataset(Enum):
 
 @unique
 class EntityKey(Enum):
+    Events = "events"
+    Sessions = "sessions"
+    Transactions = "transactions"
     MetricsSets = "metrics_sets"
     MetricsCounters = "metrics_counters"
     MetricsDistributions = "metrics_distributions"