feat(crons): Always create detectors for all monitors in MonitorValidator (#104004)

Previously, detectors were only created conditionally (when alert_rule
was present), but every cron monitor should have a detector. The
original
endpoint code from commit 450bde5 always called
ensure_cron_detector()
for all monitors.

Changes:
- Moved ensure_cron_detector() call from endpoint into
MonitorValidator.create()
- Detector creation now happens for ALL monitors (not just those with
alert_rule)
- Detector is created BEFORE alert_rule creation, allowing
IssueAlertMigrator
  to find and link it to workflows
- Added skip_detector_creation context flag to prevent duplicate
creation
  when using MonitorDataSourceValidator (new detector flow)
- Updated tests to verify detector creation

This ensures both the old (MonitorValidator) and new (detector API)
flows
correctly create detectors for all monitors.

Author: evanpurkhiser

src/sentry/monitors/endpoints/organization_monitor_index.py

@@ -46,7 +46,6 @@
     MonitorSerializer,
     MonitorSerializerResponse,
 )
-from sentry.monitors.utils import ensure_cron_detector
 from sentry.monitors.validators import MonitorBulkEditValidator, MonitorValidator
 from sentry.search.utils import tokenize_query
 from sentry.types.actor import Actor
@@ -285,7 +284,6 @@ def post(self, request: AuthenticatedHttpRequest, organization) -> Response:
             return self.respond(validator.errors, status=400)
 
         monitor = validator.save()
-        ensure_cron_detector(monitor)
         return self.respond(serialize(monitor, request.user), status=201)
 
     @extend_schema(

src/sentry/monitors/utils.py

@@ -392,7 +392,9 @@ def update_issue_alert_rule(
     return issue_alert_rule.id
 
 
-def ensure_cron_detector(monitor: Monitor):
+def ensure_cron_detector(monitor: Monitor) -> Detector | None:
+    from sentry.monitors.grouptype import MonitorIncidentType
+
     try:
         with atomic_transaction(using=router.db_for_write(DataSource)):
             data_source, created = DataSource.objects.get_or_create(
@@ -401,8 +403,6 @@ def ensure_cron_detector(monitor: Monitor):
                 source_id=str(monitor.id),
             )
             if created:
-                from sentry.monitors.grouptype import MonitorIncidentType
-
                 detector = Detector.objects.create(
                     type=MonitorIncidentType.slug,
                     project_id=monitor.project_id,
@@ -412,9 +412,19 @@ def ensure_cron_detector(monitor: Monitor):
                     config={},
                 )
                 DataSourceDetector.objects.create(data_source=data_source, detector=detector)
+                return detector
+            else:
+                return Detector.objects.get(
+                    type=MonitorIncidentType.slug,
+                    project_id=monitor.project_id,
+                    data_sources=data_source,
+                )
+
     except Exception:
         logger.exception("Error creating cron detector")
 
+    return None
+
 
 def ensure_cron_detector_deletion(monitor: Monitor):
     with atomic_transaction(using=router.db_for_write(DataSource)):

src/sentry/monitors/validators.py

@@ -41,6 +41,7 @@
 from sentry.monitors.types import CrontabSchedule, slugify_monitor_slug
 from sentry.monitors.utils import (
     create_issue_alert_rule,
+    ensure_cron_detector,
     get_checkin_margin,
     get_max_runtime,
     signal_monitor_created,
@@ -373,16 +374,22 @@ def create(self, validated_data):
             config=validated_data["config"],
         )
 
-        # Skip quota operations if requested by context (e.g., detector flow handles this)
-        if not self.context.get("skip_quota", False):
+        # When called from the new detector flow, skip detector and quota operations
+        # since they're handled at a higher level by the detector validator
+        from_detector_flow = self.context.get("from_detector_flow", False)
+
+        if not from_detector_flow:
+            detector = ensure_cron_detector(monitor)
+            assert detector
+
             # Attempt to assign a seat for this monitor
             seat_outcome = quotas.backend.assign_seat(DataCategory.MONITOR_SEAT, monitor)
             if seat_outcome != Outcome.ACCEPTED:
+                detector.update(enabled=False)
                 monitor.update(status=ObjectStatus.DISABLED)
 
         request = self.context["request"]
         signal_monitor_created(project, request.user, False, monitor, request)
-
         validated_issue_alert_rule = validated_data.get("alert_rule")
         if validated_issue_alert_rule:
             issue_alert_rule_id = create_issue_alert_rule(
@@ -643,12 +650,9 @@ def validate(self, attrs: dict[str, Any]) -> dict[str, Any]:
         if self.instance:
             monitor_instance = self.instance
 
-        # Skip quota operations - the detector validator handles seat assignment
-        context = {**self.context, "skip_quota": True}
-
         monitor_validator = MonitorValidator(
             data=monitor_data,
-            context=context,
+            context={**self.context, "from_detector_flow": True},
             instance=monitor_instance,
             partial=self.partial,
         )

tests/sentry/monitors/endpoints/test_organization_monitor_index.py

@@ -589,6 +589,10 @@ def test_simple_with_alert_rule(self) -> None:
         assert rule is not None
         assert rule.environment_id == self.environment.id
 
+        # Verify the detector was created
+        detector = get_detector_for_monitor(monitor)
+        assert detector is not None
+
     def test_checkin_margin_zero(self) -> None:
         # Invalid checkin margin
         #

tests/sentry/monitors/test_utils.py

@@ -2,7 +2,6 @@
 
 from django.db import IntegrityError
 
-from sentry.monitors.grouptype import MonitorIncidentType
 from sentry.monitors.types import DATA_SOURCE_CRON_MONITOR
 from sentry.monitors.utils import (
     ensure_cron_detector,
@@ -20,8 +19,7 @@ def setUp(self) -> None:
 
     def test_creates_data_source_and_detector_for_new_monitor(self) -> None:
         assert not get_detector_for_monitor(self.monitor)
-        ensure_cron_detector(self.monitor)
-        detector = get_detector_for_monitor(self.monitor)
+        detector = ensure_cron_detector(self.monitor)
         assert detector is not None
         assert detector.type == "monitor_check_in_failure"
         assert detector.project_id == self.monitor.project_id
@@ -30,32 +28,23 @@ def test_creates_data_source_and_detector_for_new_monitor(self) -> None:
         assert detector.owner_team_id == self.monitor.owner_team_id
 
     def test_idempotent_for_existing_data_source(self) -> None:
-        ensure_cron_detector(self.monitor)
-        detector = get_detector_for_monitor(self.monitor)
-        assert detector
-        ensure_cron_detector(self.monitor)
-        detector_after = get_detector_for_monitor(self.monitor)
+        detector = ensure_cron_detector(self.monitor)
+        assert detector is not None
+        detector_after = ensure_cron_detector(self.monitor)
         assert detector_after is not None
         assert detector.id == detector_after.id
 
     def test_with_owner_user(self) -> None:
         self.monitor.owner_user_id = self.user.id
         self.monitor.save()
-        ensure_cron_detector(self.monitor)
-        detector = Detector.objects.get(
-            type=MonitorIncidentType.slug,
-            project_id=self.monitor.project_id,
-        )
+        detector = ensure_cron_detector(self.monitor)
+        assert detector is not None
         assert detector.owner_user_id == self.user.id
         assert detector.owner_team_id is None
 
     def test_with_no_owner(self) -> None:
-        ensure_cron_detector(self.monitor)
-
-        detector = Detector.objects.get(
-            type=MonitorIncidentType.slug,
-            project_id=self.monitor.project_id,
-        )
+        detector = ensure_cron_detector(self.monitor)
+        assert detector is not None
         assert detector.owner_user_id is None
         assert detector.owner_team_id is None
 

tests/sentry/monitors/test_validators.py

@@ -19,6 +19,7 @@
     is_monitor_muted,
 )
 from sentry.monitors.types import DATA_SOURCE_CRON_MONITOR
+from sentry.monitors.utils import get_detector_for_monitor
 from sentry.monitors.validators import (
     MonitorDataSourceValidator,
     MonitorIncidentDetectorValidator,
@@ -262,6 +263,11 @@ def test_create_monitor_without_seat(self, assign_seat):
         monitor.refresh_from_db()
         assert monitor.status == ObjectStatus.DISABLED
 
+        # Verify the detector is also disabled when quota is exceeded
+        detector = get_detector_for_monitor(monitor)
+        assert detector is not None
+        assert detector.enabled is False
+
     def test_invalid_schedule(self):
         data = {
             "project": self.project.slug,