feat: Add anomaly detection for log alerts (#95244)

Add anomaly detection support for log alerts. Uses dataset label + event
type to determine when to query logs.
Removes timestamp rounding now that the timeseries API handles stable
quantization.

Author: shruthilayaj

src/sentry/seer/anomaly_detection/store_data.py

@@ -24,7 +24,7 @@
 from sentry.seer.anomaly_detection.utils import (
     fetch_historical_data,
     format_historical_data,
-    get_dataset_from_label,
+    get_dataset_from_label_and_event_types,
     get_event_types,
     translate_direction,
 )
@@ -156,9 +156,9 @@ def send_historical_data_to_seer(
     if not snuba_query:
         snuba_query = SnubaQuery.objects.get(id=alert_rule.snuba_query_id)
     window_min = int(snuba_query.time_window / 60)
-    dataset = get_dataset_from_label(snuba_query.dataset)
-    query_columns = get_query_columns([snuba_query.aggregate], window_min)
     event_types = get_event_types(snuba_query, event_types)
+    dataset = get_dataset_from_label_and_event_types(snuba_query.dataset, event_types)
+    query_columns = get_query_columns([snuba_query.aggregate], window_min)
     if not alert_rule.organization:
         raise ValidationError("Alert rule doesn't belong to an organization")
 

src/sentry/seer/anomaly_detection/utils.py

@@ -1,4 +1,4 @@
-from datetime import UTC, datetime, timedelta
+from datetime import datetime, timedelta
 from typing import Any
 
 from django.utils import timezone
@@ -14,7 +14,7 @@
 from sentry.search.eap.types import SearchResolverConfig
 from sentry.search.events.types import SnubaParams
 from sentry.seer.anomaly_detection.types import AnomalyType, TimeSeriesPoint
-from sentry.snuba import metrics_performance, spans_rpc
+from sentry.snuba import metrics_performance, ourlogs, spans_rpc
 from sentry.snuba.metrics.extraction import MetricSpecType
 from sentry.snuba.models import SnubaQuery, SnubaQueryEventType
 from sentry.snuba.referrer import Referrer
@@ -229,12 +229,17 @@ def format_historical_data(
     )
 
 
-def get_dataset_from_label(dataset_label: str):
+def get_dataset_from_label_and_event_types(
+    dataset_label: str, event_types: list[SnubaQueryEventType.EventType] | None = None
+):
     if dataset_label == "events":
         # DATASET_OPTIONS expects the name 'errors'
         dataset_label = "errors"
     elif dataset_label == "events_analytics_platform":
-        dataset_label = "spans"
+        if event_types and SnubaQueryEventType.EventType.TRACE_ITEM_LOG in event_types:
+            dataset_label = "ourlogs"
+        else:
+            dataset_label = "spans"
     elif dataset_label in ["generic_metrics", "transactions"]:
         # XXX: performance alerts dataset differs locally vs in prod
         dataset_label = "metricsEnhanced"
@@ -266,8 +271,8 @@ def fetch_historical_data(
     if start is None:
         start = end - timedelta(days=NUM_DAYS)
     granularity = snuba_query.time_window
-
-    dataset = get_dataset_from_label(snuba_query.dataset)
+    event_types = get_event_types(snuba_query, event_types)
+    dataset = get_dataset_from_label_and_event_types(snuba_query.dataset, event_types)
 
     if not project or not dataset or not organization:
         return None
@@ -289,16 +294,6 @@ def fetch_historical_data(
     if dataset == metrics_performance:
         return get_crash_free_historical_data(start, end, project, organization, granularity)
     elif dataset == spans_rpc:
-        # EAP timeseries don't round time buckets to the nearest time window but seer expects
-        # that. So for example, if start was 7:01 with a 15 min interval, EAP would
-        # bucket it as 7:01, 7:16 etc. Force rounding the start and end times so we
-        # get the buckets seer expects.
-        rounded_end = int(end.timestamp() / granularity) * granularity
-        rounded_start = int(start.timestamp() / granularity) * granularity
-
-        snuba_params.end = datetime.fromtimestamp(rounded_end, UTC)
-        snuba_params.start = datetime.fromtimestamp(rounded_start, UTC)
-
         results = spans_rpc.run_timeseries_query(
             params=snuba_params,
             query_string=snuba_query.query,
@@ -315,8 +310,24 @@ def fetch_historical_data(
             sampling_mode="NORMAL",
         )
         return results
+    elif dataset == ourlogs:
+        results = ourlogs.run_timeseries_query(
+            params=snuba_params,
+            query_string=snuba_query.query,
+            y_axes=query_columns,
+            referrer=(
+                Referrer.ANOMALY_DETECTION_HISTORICAL_DATA_QUERY.value
+                if is_store_data_request
+                else Referrer.ANOMALY_DETECTION_RETURN_HISTORICAL_ANOMALIES.value
+            ),
+            config=SearchResolverConfig(
+                auto_fields=False,
+                use_aggregate_conditions=False,
+            ),
+            sampling_mode="NORMAL",
+        )
+        return results
     else:
-        event_types = get_event_types(snuba_query, event_types)
         snuba_query_string = get_snuba_query_string(snuba_query, event_types)
         historical_data = dataset.timeseries_query(
             selected_columns=query_columns,

tests/sentry/incidents/endpoints/test_organization_alert_rule_index.py

@@ -49,6 +49,8 @@
 from sentry.snuba.dataset import Dataset
 from sentry.snuba.metrics.naming_layer.mri import SessionMRI
 from sentry.snuba.models import SnubaQueryEventType
+from sentry.snuba.ourlogs import run_timeseries_query as ourlogs_run_timeseries_query
+from sentry.snuba.spans_rpc import run_timeseries_query as spans_rpc_run_timeseries_query
 from sentry.snuba.tasks import create_subscription_in_snuba
 from sentry.testutils.abstract import Abstract
 from sentry.testutils.cases import APITestCase, SnubaTestCase
@@ -393,7 +395,13 @@ def test_anomaly_detection_alert(self, mock_seer_request):
     @patch(
         "sentry.seer.anomaly_detection.store_data.seer_anomaly_detection_connection_pool.urlopen"
     )
-    def test_anomaly_detection_alert_eap(self, mock_seer_request):
+    @patch(
+        "sentry.seer.anomaly_detection.utils.spans_rpc.run_timeseries_query",
+        wraps=spans_rpc_run_timeseries_query,
+    )
+    def test_anomaly_detection_alert_eap_spans(
+        self, mock_spans_timeseries_query, mock_seer_request
+    ):
         data = deepcopy(self.dynamic_alert_rule_dict)
         data["dataset"] = "events_analytics_platform"
         data["alertType"] = "eap_metrics"
@@ -412,6 +420,42 @@ def test_anomaly_detection_alert_eap(self, mock_seer_request):
         assert alert_rule.seasonality == resp.data.get("seasonality")
         assert alert_rule.sensitivity == resp.data.get("sensitivity")
         assert mock_seer_request.call_count == 1
+        assert mock_spans_timeseries_query.call_count == 1
+
+    @with_feature("organizations:anomaly-detection-alerts")
+    @with_feature("organizations:anomaly-detection-rollout")
+    @with_feature("organizations:ourlogs-alerts")
+    @with_feature("organizations:incidents")
+    @patch(
+        "sentry.seer.anomaly_detection.store_data.seer_anomaly_detection_connection_pool.urlopen"
+    )
+    @patch(
+        "sentry.seer.anomaly_detection.utils.ourlogs.run_timeseries_query",
+        wraps=ourlogs_run_timeseries_query,
+    )
+    def test_anomaly_detection_alert_ourlogs(
+        self, mock_ourlogs_run_timeseries_query, mock_seer_request
+    ):
+        data = deepcopy(self.dynamic_alert_rule_dict)
+        data["dataset"] = "events_analytics_platform"
+        data["alertType"] = "trace_item_logs"
+        data["eventTypes"] = ["trace_item_log"]
+        seer_return_value: StoreDataResponse = {"success": True}
+        mock_seer_request.return_value = HTTPResponse(orjson.dumps(seer_return_value), status=200)
+
+        with outbox_runner():
+            resp = self.get_success_response(
+                self.organization.slug,
+                status_code=201,
+                **data,
+            )
+        assert "id" in resp.data
+        alert_rule = AlertRule.objects.get(id=resp.data["id"])
+        assert resp.data == serialize(alert_rule, self.user)
+        assert alert_rule.seasonality == resp.data.get("seasonality")
+        assert alert_rule.sensitivity == resp.data.get("sensitivity")
+        assert mock_seer_request.call_count == 1
+        assert mock_ourlogs_run_timeseries_query.call_count == 1
 
     @patch(
         "sentry.snuba.subscriptions.create_subscription_in_snuba.delay",