feat(investigations): Read and build monitor changes for update_detector

src/sentry/investigations/services/action_kinds/update_detector.py

@@ -1,10 +1,13 @@
 from __future__ import annotations
 
-from typing import Any
+from dataclasses import dataclass
+from typing import Any, Literal, TypedDict, cast
 
 from rest_framework import serializers
 
+from sentry.api.serializers import serialize
 from sentry.db.models.fields.bounded import I32_MAX, I64_MAX
+from sentry.incidents.endpoints.serializers.query_subscription import SnubaQuerySerializer
 from sentry.investigations.contracts import (
     ActionPayloadSerializer,
     StrictCharField,
@@ -13,6 +16,12 @@
     StrictIntegerField,
 )
 from sentry.investigations.services.action_kinds.base import ActionKind
+from sentry.snuba.models import SnubaQuery
+from sentry.workflow_engine.endpoints.serializers.data_condition_serializer import (
+    DataConditionSerializer,
+)
+from sentry.workflow_engine.models import Detector
+from sentry.workflow_engine.types import DetectorPriorityLevel
 
 UPDATE_DETECTOR_ACTION = "update_detector"
 
@@ -96,6 +105,153 @@ def trim(cls, action: dict[str, Any]) -> dict[str, Any] | None:
         }
 
 
+DetectorValue = int | float | str | None
+
+_THRESHOLD_RESULTS: dict[
+    Literal["highThreshold", "mediumThreshold", "resolutionThreshold"], DetectorPriorityLevel
+] = {
+    "highThreshold": DetectorPriorityLevel.HIGH,
+    "mediumThreshold": DetectorPriorityLevel.MEDIUM,
+    "resolutionThreshold": DetectorPriorityLevel.OK,
+}
+_DATA_SOURCE_KEYS = ("timeWindowSeconds", "query", "aggregate", "environment")
+
+
+class DetectorChanges(TypedDict, total=False):
+    timeWindowSeconds: int
+    highThreshold: float
+    mediumThreshold: float
+    resolutionThreshold: float
+    query: str
+    aggregate: str
+    environment: str | None
+
+
+class DataSourceBody(TypedDict):
+    id: str
+    dataset: str
+    query: str
+    aggregate: str
+    timeWindow: int
+    environment: str | None
+    eventTypes: list[str]
+    queryType: int
+
+
+class ConditionBody(TypedDict):
+    id: str
+    type: str
+    # Only static monitors get here, and their conditions compare against a number.
+    comparison: float
+    conditionResult: int
+
+
+class ConditionGroupBody(TypedDict):
+    id: int
+    logicType: str
+    conditions: list[ConditionBody]
+
+
+class DetectorUpdate(TypedDict, total=False):
+    """The body we send to the detector validator."""
+
+    conditionGroup: ConditionGroupBody
+    dataSources: list[DataSourceBody]
+
+
+@dataclass(frozen=True)
+class DetectorState:
+    """A metric monitor's query and conditions, serialized the way the detector API returns
+    them, so we can read current values and build an update from the same data.
+    """
+
+    data_source: DataSourceBody
+    condition_group: ConditionGroupBody | None
+
+    @classmethod
+    def load(cls, detector: Detector, snuba_query: SnubaQuery) -> DetectorState:
+        serialized = serialize(snuba_query, serializer=SnubaQuerySerializer())
+        # Without a mode the update keeps the stored one. Some span monitors migrated from
+        # transactions store a mode the validator no longer accepts.
+        del serialized["extrapolationMode"]
+        data_source = cast(DataSourceBody, {**serialized, "queryType": snuba_query.type})
+        group = detector.workflow_condition_group
+        condition_group = (
+            ConditionGroupBody(
+                id=group.id,
+                logicType=group.logic_type,
+                conditions=cast(
+                    list[ConditionBody],
+                    serialize(
+                        list(group.conditions.order_by("id")), serializer=DataConditionSerializer()
+                    ),
+                ),
+            )
+            if group is not None
+            else None
+        )
+        return cls(data_source=data_source, condition_group=condition_group)
+
+    def values(self) -> dict[str, DetectorValue]:
+        """The monitor's current settings, keyed like Seer's changes. A threshold is left out
+        unless exactly one condition sets it.
+        """
+
+        values: dict[str, DetectorValue] = {
+            "timeWindowSeconds": self.data_source["timeWindow"],
+            "query": self.data_source["query"],
+            "aggregate": self.data_source["aggregate"],
+            "environment": self.data_source["environment"],
+        }
+        conditions = self.condition_group["conditions"] if self.condition_group else []
+        for key, result in _THRESHOLD_RESULTS.items():
+            comparisons = [
+                condition["comparison"]
+                for condition in conditions
+                if condition["conditionResult"] == result
+            ]
+            if len(comparisons) == 1:
+                values[key] = comparisons[0]
+        return values
+
+    def update(self, changes: DetectorChanges) -> DetectorUpdate:
+        """Apply Seer's changes on top of the current settings."""
+
+        update: DetectorUpdate = {}
+        thresholds: dict[int, float] = {
+            result: changes[key] for key, result in _THRESHOLD_RESULTS.items() if key in changes
+        }
+        if thresholds:
+            # values() has no thresholds without a condition group, so the stale check stops
+            # those changes before they get here.
+            assert self.condition_group is not None
+            # The update deletes conditions we leave out, so send them all.
+            update["conditionGroup"] = {
+                **self.condition_group,
+                "conditions": [
+                    {
+                        **condition,
+                        "comparison": thresholds.get(
+                            condition["conditionResult"], condition["comparison"]
+                        ),
+                    }
+                    for condition in self.condition_group["conditions"]
+                ],
+            }
+        if any(key in changes for key in _DATA_SOURCE_KEYS):
+            data_source = self.data_source.copy()
+            if "timeWindowSeconds" in changes:
+                data_source["timeWindow"] = changes["timeWindowSeconds"]
+            if "query" in changes:
+                data_source["query"] = changes["query"]
+            if "aggregate" in changes:
+                data_source["aggregate"] = changes["aggregate"]
+            if "environment" in changes:
+                data_source["environment"] = changes["environment"]
+            update["dataSources"] = [data_source]
+        return update
+
+
 class UpdateDetectorKind(ActionKind):
     kind = UPDATE_DETECTOR_ACTION
     payload_serializer = UpdateDetectorActionSerializer

tests/sentry/investigations/services/test_update_detector_changes.py

@@ -0,0 +1,166 @@
+from __future__ import annotations
+
+from datetime import timedelta
+from typing import Any
+
+from sentry.incidents.grouptype import MetricIssue
+from sentry.incidents.metric_issue_detector import fetch_snuba_query
+from sentry.incidents.utils.constants import INCIDENTS_SNUBA_SUBSCRIPTION_TYPE
+from sentry.incidents.utils.subscription_limits import METRIC_SUBSCRIPTION_FEATURE_FLAGS
+from sentry.investigations.services.action_kinds.update_detector import (
+    DetectorChanges,
+    DetectorState,
+)
+from sentry.snuba.dataset import Dataset
+from sentry.snuba.models import ExtrapolationMode, SnubaQuery, SnubaQueryEventType
+from sentry.snuba.subscriptions import create_snuba_query, create_snuba_subscription
+from sentry.testutils.cases import TestCase
+from sentry.testutils.helpers.features import with_feature
+from sentry.testutils.skips import requires_kafka, requires_snuba
+from sentry.workflow_engine.models import DataConditionGroup, Detector
+from sentry.workflow_engine.models.data_condition import Condition
+from sentry.workflow_engine.types import DetectorPriorityLevel
+
+pytestmark = [requires_snuba, requires_kafka]
+
+
+@with_feature(METRIC_SUBSCRIPTION_FEATURE_FLAGS)
+class DetectorStateTest(TestCase):
+    def setUp(self) -> None:
+        super().setUp()
+        self.detector = self.create_metric_detector()
+
+    def create_metric_detector(
+        self,
+        *,
+        eap: bool = False,
+        extrapolation_mode: ExtrapolationMode = ExtrapolationMode.CLIENT_AND_SERVER_WEIGHTED,
+    ) -> Detector:
+        with self.tasks():
+            snuba_query = create_snuba_query(
+                query_type=SnubaQuery.Type.PERFORMANCE if eap else SnubaQuery.Type.ERROR,
+                dataset=Dataset.EventsAnalyticsPlatform if eap else Dataset.Events,
+                query="span.op:http.client" if eap else "level:error",
+                aggregate="count(span.duration)" if eap else "count()",
+                time_window=timedelta(minutes=5),
+                resolution=timedelta(minutes=1),
+                environment=None,
+                event_types=[
+                    SnubaQueryEventType.EventType.TRACE_ITEM_SPAN
+                    if eap
+                    else SnubaQueryEventType.EventType.ERROR
+                ],
+                extrapolation_mode=extrapolation_mode if eap else None,
+            )
+            subscription = create_snuba_subscription(
+                project=self.project,
+                subscription_type=INCIDENTS_SNUBA_SUBSCRIPTION_TYPE,
+                snuba_query=snuba_query,
+            )
+        group = self.create_data_condition_group(
+            organization_id=self.organization.id, logic_type=DataConditionGroup.Type.ANY
+        )
+        for condition_type, result in [
+            (Condition.GREATER, DetectorPriorityLevel.HIGH),
+            (Condition.LESS_OR_EQUAL, DetectorPriorityLevel.OK),
+        ]:
+            self.create_data_condition(
+                condition_group=group,
+                type=condition_type,
+                comparison=500.0,
+                condition_result=result,
+            )
+        detector = self.create_detector(
+            project=self.project,
+            type=MetricIssue.slug,
+            workflow_condition_group=group,
+            config={"detection_type": "static"},
+        )
+        data_source = self.create_data_source(
+            organization=self.organization, source_id=subscription.id
+        )
+        self.create_data_source_detector(data_source=data_source, detector=detector)
+        return detector
+
+    def state(self) -> DetectorState:
+        snuba_query = fetch_snuba_query(self.detector)
+        assert snuba_query is not None
+        return DetectorState.load(self.detector, snuba_query)
+
+    def validation_errors(self, changes: DetectorChanges) -> dict[str, Any] | None:
+        """Send the update through the detector validator, the way the detector API would."""
+        settings = self.detector.group_type.detector_settings
+        assert settings is not None and settings.validator is not None
+        validator = settings.validator(
+            instance=self.detector,
+            context={"project": self.project, "organization": self.organization},
+            data=self.state().update(changes),
+            partial=True,
+        )
+        return None if validator.is_valid() else validator.errors
+
+    def test_reads_the_current_values(self) -> None:
+        assert self.state().values() == {
+            "timeWindowSeconds": 300,
+            "query": "level:error",
+            "aggregate": "count()",
+            "environment": None,
+            "highThreshold": 500.0,
+            "resolutionThreshold": 500.0,
+        }
+
+    def test_a_valid_change_has_no_errors(self) -> None:
+        assert self.validation_errors({"timeWindowSeconds": 900}) is None
+
+    def test_a_threshold_change_passes_the_validator(self) -> None:
+        assert (
+            self.validation_errors({"highThreshold": 800.0, "resolutionThreshold": 800.0}) is None
+        )
+
+    def test_a_threshold_change_updates_its_condition_and_keeps_the_rest(self) -> None:
+        update = self.state().update({"highThreshold": 800.0})
+
+        assert update.keys() == {"conditionGroup"}
+        conditions = update["conditionGroup"]["conditions"]
+        assert [(c["conditionResult"], c["comparison"]) for c in conditions] == [
+            (DetectorPriorityLevel.HIGH, 800.0),
+            (DetectorPriorityLevel.OK, 500.0),
+        ]
+
+    def test_a_time_window_change_only_touches_the_data_source(self) -> None:
+        update = self.state().update({"timeWindowSeconds": 900})
+
+        assert update.keys() == {"dataSources"}
+        [data_source] = update["dataSources"]
+        assert data_source["timeWindow"] == 900
+        assert data_source["query"] == "level:error"
+
+    def test_a_group_by_monitor_can_still_change(self) -> None:
+        snuba_query = fetch_snuba_query(self.detector)
+        assert snuba_query is not None
+        snuba_query.update(group_by=["transaction"])
+
+        assert "groupBy" not in self.state().update({"timeWindowSeconds": 900})["dataSources"][0]
+        assert self.validation_errors({"timeWindowSeconds": 900}) is None
+
+    def test_a_span_monitor_can_change_its_window(self) -> None:
+        self.detector = self.create_metric_detector(eap=True)
+
+        assert self.validation_errors({"timeWindowSeconds": 900}) is None
+
+    def test_a_migrated_span_monitor_keeps_its_extrapolation_mode(self) -> None:
+        self.detector = self.create_metric_detector(
+            eap=True, extrapolation_mode=ExtrapolationMode.SERVER_WEIGHTED
+        )
+
+        [data_source] = self.state().update({"timeWindowSeconds": 900})["dataSources"]
+
+        assert "extrapolationMode" not in data_source
+        assert self.validation_errors({"timeWindowSeconds": 900}) is None
+
+    def test_an_invalid_window_returns_the_validator_errors(self) -> None:
+        self.detector = self.create_metric_detector(eap=True)
+
+        errors = self.validation_errors({"timeWindowSeconds": 450})
+
+        assert "Invalid Time Window" in str(errors)