feat(alerts): Add support for log type alert

src/sentry/snuba/entity_subscription.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import re
+import types
 from abc import ABC, abstractmethod
 from collections.abc import Mapping, MutableMapping, Sequence
 from dataclasses import dataclass
@@ -23,7 +24,7 @@
 from sentry.search.events.types import ParamsType, QueryBuilderConfig, SnubaParams
 from sentry.sentry_metrics.use_case_id_registry import UseCaseID
 from sentry.sentry_metrics.utils import resolve, resolve_tag_key, resolve_tag_values
-from sentry.snuba import rpc_dataset_common, spans_rpc
+from sentry.snuba import ourlogs, rpc_dataset_common, spans_rpc
 from sentry.snuba.dataset import Dataset, EntityKey
 from sentry.snuba.metrics.extraction import MetricSpecType
 from sentry.snuba.metrics.naming_layer.mri import SessionMRI
@@ -265,6 +266,11 @@ def build_rpc_request(
         if environment:
             params["environment"] = environment.name
 
+        dataset_module: types.ModuleType
+        if self.event_types and self.event_types[0] == SnubaQueryEventType.EventType.TRACE_ITEM_LOG:
+            dataset_module = ourlogs
+        else:
+            dataset_module = spans_rpc
         now = datetime.now(tz=timezone.utc)
         snuba_params = SnubaParams(
             environments=[environment],
@@ -274,7 +280,7 @@ def build_rpc_request(
             end=now,
             granularity_secs=self.time_window,
         )
-        search_resolver = spans_rpc.get_resolver(
+        search_resolver = dataset_module.get_resolver(
             snuba_params, SearchResolverConfig(stable_timestamp_quantization=False)
         )
 

src/sentry/snuba/models.py

@@ -81,6 +81,8 @@ class EventType(Enum):
         ERROR = 0
         DEFAULT = 1
         TRANSACTION = 2
+        TRACE_ITEM_SPAN = 3
+        TRACE_ITEM_LOG = 4
 
     snuba_query = FlexibleForeignKey("sentry.SnubaQuery")
     type = models.SmallIntegerField()

src/sentry/snuba/snuba_query_validator.py

@@ -61,7 +61,11 @@
         SnubaQueryEventType.EventType.ERROR,
         SnubaQueryEventType.EventType.DEFAULT,
     },
-    SnubaQuery.Type.PERFORMANCE: {SnubaQueryEventType.EventType.TRANSACTION},
+    SnubaQuery.Type.PERFORMANCE: {
+        SnubaQueryEventType.EventType.TRANSACTION,
+        SnubaQueryEventType.EventType.TRACE_ITEM_LOG,
+        SnubaQueryEventType.EventType.TRACE_ITEM_SPAN,
+    },
 }
 
 
@@ -119,13 +123,22 @@ def validate_query(self, query: str):
 
     def validate_event_types(self, value: Sequence[str]) -> list[SnubaQueryEventType.EventType]:
         try:
-            return [SnubaQueryEventType.EventType[event_type.upper()] for event_type in value]
+            validated = [SnubaQueryEventType.EventType[event_type.upper()] for event_type in value]
         except KeyError:
             raise serializers.ValidationError(
                 "Invalid event_type, valid values are %s"
                 % [item.name.lower() for item in SnubaQueryEventType.EventType]
             )
 
+        if not features.has(
+            "organizations:ourlogs-alerts",
+            self.context["organization"],
+            actor=self.context.get("user", None),
+        ) and any([v for v in validated if v == SnubaQueryEventType.EventType.TRACE_ITEM_LOG]):
+            raise serializers.ValidationError("You do not have access to the log alerts feature.")
+
+        return validated
+
     def validate(self, data):
         data = super().validate(data)
         self._validate_aggregate(data)
@@ -143,6 +156,13 @@ def validate(self, data):
                 % sorted(et.name.lower() for et in valid_event_types)
             )
 
+        dataset = data.get("dataset")
+        if dataset == Dataset.EventsAnalyticsPlatform and event_types and len(event_types) > 1:
+            raise serializers.ValidationError(
+                "Multiple event types not allowed. Valid event types are %s"
+                % sorted(et.name.lower() for et in valid_event_types)
+            )
+
         return data
 
     def _validate_aggregate(self, data):

tests/sentry/incidents/endpoints/test_organization_alert_rule_index.py

@@ -11,6 +11,8 @@
 from httpx import HTTPError
 from rest_framework import status
 from rest_framework.exceptions import ErrorDetail
+from sentry_protos.snuba.v1.endpoint_create_subscription_pb2 import CreateSubscriptionRequest
+from sentry_protos.snuba.v1.request_common_pb2 import TraceItemType
 from urllib3.exceptions import MaxRetryError, TimeoutError
 from urllib3.response import HTTPResponse
 
@@ -46,6 +48,8 @@
 from sentry.silo.base import SiloMode
 from sentry.snuba.dataset import Dataset
 from sentry.snuba.metrics.naming_layer.mri import SessionMRI
+from sentry.snuba.models import SnubaQueryEventType
+from sentry.snuba.tasks import create_subscription_in_snuba
 from sentry.testutils.abstract import Abstract
 from sentry.testutils.cases import APITestCase, SnubaTestCase
 from sentry.testutils.factories import EventType
@@ -54,6 +58,7 @@
 from sentry.testutils.outbox import outbox_runner
 from sentry.testutils.silo import assume_test_silo_mode
 from sentry.testutils.skips import requires_snuba
+from sentry.utils.snuba import _snuba_pool
 from sentry.workflow_engine.models import (
     Action,
     ActionAlertRuleTriggerAction,
@@ -408,23 +413,89 @@ def test_anomaly_detection_alert_eap(self, mock_seer_request):
         assert alert_rule.sensitivity == resp.data.get("sensitivity")
         assert mock_seer_request.call_count == 1
 
-    def test_create_alert_rule_eap(self):
+    @patch(
+        "sentry.snuba.subscriptions.create_subscription_in_snuba.delay",
+        wraps=create_subscription_in_snuba,
+    )
+    def test_create_alert_rule_eap_spans(self, mock_create_subscription_in_snuba):
+        for event_type in ["transaction", "trace_item_span", None]:
+            data = deepcopy(self.alert_rule_dict)
+            data["dataset"] = "events_analytics_platform"
+            data["alertType"] = "eap_metrics"
+            data["timeWindow"] = 5
+            if event_type:
+                data["eventTypes"] = [event_type]
+
+            with (
+                outbox_runner(),
+                self.feature(["organizations:incidents", "organizations:performance-view"]),
+            ):
+                with patch.object(
+                    _snuba_pool, "urlopen", side_effect=_snuba_pool.urlopen
+                ) as urlopen:
+                    resp = self.get_success_response(
+                        self.organization.slug,
+                        status_code=201,
+                        **data,
+                    )
+
+                    rpc_request_body = urlopen.call_args[1]["body"]
+                    createSubscriptionRequest = CreateSubscriptionRequest.FromString(
+                        rpc_request_body
+                    )
+
+                    assert (
+                        createSubscriptionRequest.time_series_request.meta.trace_item_type
+                        == TraceItemType.TRACE_ITEM_TYPE_SPAN
+                    )
+
+            assert "id" in resp.data
+            alert_rule = AlertRule.objects.get(id=resp.data["id"])
+            assert resp.data == serialize(alert_rule, self.user)
+
+    @patch(
+        "sentry.snuba.subscriptions.create_subscription_in_snuba.delay",
+        wraps=create_subscription_in_snuba,
+    )
+    def test_create_alert_rule_logs(self, mock_create_subscription_in_snuba):
         data = deepcopy(self.alert_rule_dict)
         data["dataset"] = "events_analytics_platform"
         data["alertType"] = "eap_metrics"
-        data["aggregate"] = "count_unique(org)"
+        data["aggregate"] = "count(message)"
+        data["eventTypes"] = ["trace_item_log"]
+        data["timeWindow"] = 5
         with (
             outbox_runner(),
-            self.feature(["organizations:incidents", "organizations:performance-view"]),
+            self.feature(
+                [
+                    "organizations:incidents",
+                    "organizations:performance-view",
+                    "organizations:ourlogs-alerts",
+                ]
+            ),
         ):
-            resp = self.get_success_response(
-                self.organization.slug,
-                status_code=201,
-                **data,
-            )
+            with patch.object(_snuba_pool, "urlopen", side_effect=_snuba_pool.urlopen) as urlopen:
+                resp = self.get_success_response(
+                    self.organization.slug,
+                    status_code=201,
+                    **data,
+                )
+
+                rpc_request_body = urlopen.call_args[1]["body"]
+                createSubscriptionRequest = CreateSubscriptionRequest.FromString(rpc_request_body)
+
+                assert (
+                    createSubscriptionRequest.time_series_request.meta.trace_item_type
+                    == TraceItemType.TRACE_ITEM_TYPE_LOG
+                )
+
         assert "id" in resp.data
         alert_rule = AlertRule.objects.get(id=resp.data["id"])
         assert resp.data == serialize(alert_rule, self.user)
+        assert (
+            SnubaQueryEventType.objects.filter(snuba_query_id=alert_rule.snuba_query_id)[0].type
+            == SnubaQueryEventType.EventType.TRACE_ITEM_LOG.value
+        )
 
     @with_feature("organizations:anomaly-detection-alerts")
     @with_feature("organizations:anomaly-detection-rollout")