feat(crons): Check quotas.check_accept_monitor_checkin in monitor_con…

…sumer (#62512)

Checks that a monitor check-in can be accepted via the quotas system.

In Sentry SaaS this will be driven by the billing seat system, producing
PermitCheckInStatus.DROP when a monitor has not been assigned a seat.

We do this as a performance optimization so we can drop check-ins for
unpaid monitors as early as possible

src/sentry/monitors/consumers/monitor_consumer.py

@@ -18,10 +18,11 @@
 from django.db import router, transaction
 from sentry_sdk.tracing import Span, Transaction
 
-from sentry import ratelimits
+from sentry import quotas, ratelimits
 from sentry.constants import DataCategory, ObjectStatus
 from sentry.killswitches import killswitch_matches_context
 from sentry.models.project import Project
+from sentry.monitors.constants import PermitCheckInStatus
 from sentry.monitors.logic.mark_failed import mark_failed
 from sentry.monitors.logic.mark_ok import mark_ok
 from sentry.monitors.models import (
@@ -58,7 +59,9 @@ def _ensure_monitor_with_config(
     project: Project,
     monitor_slug: str,
     config: Optional[Dict],
+    quotas_outcome: PermitCheckInStatus,
 ):
+
     try:
         monitor = Monitor.objects.get(
             slug=monitor_slug,
@@ -118,6 +121,13 @@ def _ensure_monitor_with_config(
     if monitor and not created and monitor.config != validated_config:
         monitor.update_config(config, validated_config)
 
+    # When accepting for upsert attempt to assign a seat for the monitor,
+    # otherwise the monitor is marked as disabled
+    if monitor and quotas_outcome == PermitCheckInStatus.ACCEPTED_FOR_UPSERT:
+        seat_outcome = quotas.backend.assign_monitor_seat(monitor)
+        if seat_outcome != Outcome.ACCEPTED:
+            monitor.update(status=ObjectStatus.DISABLED)
+
     return monitor
 
 
@@ -339,6 +349,23 @@ def _process_checkin(item: CheckinItem, txn: Transaction | Span):
         )
         return
 
+    # Does quotas allow for this check-in to be accepted?
+    quotas_outcome: PermitCheckInStatus = quotas.backend.check_accept_monitor_checkin(
+        project.id, monitor_slug
+    )
+
+    if quotas_outcome == PermitCheckInStatus.DROP:
+        track_outcome(
+            org_id=project.organization_id,
+            project_id=project.id,
+            key_id=None,
+            outcome=Outcome.RATE_LIMITED,
+            reason="over_quota",
+            timestamp=start_time,
+            category=DataCategory.MONITOR,
+        )
+        return
+
     guid, use_latest_checkin = transform_checkin_uuid(
         txn,
         metric_kwargs,
@@ -406,6 +433,7 @@ def _process_checkin(item: CheckinItem, txn: Transaction | Span):
             project,
             monitor_slug,
             monitor_config,
+            quotas_outcome,
         )
     except MonitorLimitsExceeded:
         metrics.incr(
@@ -449,7 +477,29 @@ def _process_checkin(item: CheckinItem, txn: Transaction | Span):
         )
         return
 
+    # When a monitor was accepted for upsert but is disabled we were unable to
+    # assign a seat. Discard the check-in in this case.
+    if (
+        quotas_outcome == PermitCheckInStatus.ACCEPTED_FOR_UPSERT
+        and monitor.status == ObjectStatus.DISABLED
+    ):
+        track_outcome(
+            org_id=project.organization_id,
+            project_id=project.id,
+            key_id=None,
+            outcome=Outcome.RATE_LIMITED,
+            reason="over_quota",
+            timestamp=start_time,
+            category=DataCategory.MONITOR,
+        )
+        return
+
     # Discard check-ins if the monitor is disabled
+    #
+    # Typically a disabled monitor will result in a PermitCheckInStatus.DROP
+    # and we'll have dropped the check in earlier during processing. This check
+    # is here for the on-premise version of Sentry where quotas always accepts
+    # check-ins, even when the monitor is disabled.
     if monitor.status == ObjectStatus.DISABLED:
         metrics.incr(
             "monitors.checkin.result",

tests/sentry/monitors/test_monitor_consumer.py

@@ -12,7 +12,7 @@
 from sentry import killswitches
 from sentry.constants import ObjectStatus
 from sentry.db.models import BoundedPositiveIntegerField
-from sentry.monitors.constants import TIMEOUT
+from sentry.monitors.constants import TIMEOUT, PermitCheckInStatus
 from sentry.monitors.consumers.monitor_consumer import StoreMonitorCheckInStrategyFactory
 from sentry.monitors.models import (
     CheckInStatus,
@@ -26,6 +26,7 @@
 from sentry.testutils.cases import TestCase
 from sentry.utils import json
 from sentry.utils.locking.manager import LockManager
+from sentry.utils.outcomes import Outcome
 from sentry.utils.services import build_instance_from_options
 
 locks = LockManager(build_instance_from_options(settings.SENTRY_POST_PROCESS_LOCKS_BACKEND_OPTIONS))
@@ -605,3 +606,93 @@ def test_monitor_tasks_trigger(self, try_monitor_tasks_trigger):
             assert MonitorCheckIn.objects.filter(guid=self.guid).exists()
             logger.exception.assert_called_with("Failed to trigger monitor tasks")
             try_monitor_tasks_trigger.side_effect = None
+
+    @mock.patch("sentry.quotas.backend.check_accept_monitor_checkin")
+    def test_monitor_quotas_accept(self, check_accept_monitor_checkin):
+        check_accept_monitor_checkin.return_value = PermitCheckInStatus.ACCEPT
+
+        # Explicitly leaving off the "disabled" status to validate that we're
+        # not dropping due to the monitor being disabled
+        monitor = self._create_monitor(slug="my-monitor")
+        self.send_checkin(monitor.slug)
+
+        check_accept_monitor_checkin.assert_called_with(self.project.id, monitor.slug)
+
+        checkin = MonitorCheckIn.objects.get(monitor_id=monitor.id)
+        assert checkin.status == CheckInStatus.OK
+
+    @mock.patch("sentry.quotas.backend.check_accept_monitor_checkin")
+    def test_monitor_quotas_drop(self, check_accept_monitor_checkin):
+        check_accept_monitor_checkin.return_value = PermitCheckInStatus.DROP
+
+        # Explicitly leaving off the "disabled" status to validate that we're
+        # not dropping due to the monitor being disabled
+        monitor = self._create_monitor(slug="my-monitor")
+        self.send_checkin(monitor.slug)
+
+        check_accept_monitor_checkin.assert_called_with(self.project.id, monitor.slug)
+
+        checkins = MonitorCheckIn.objects.filter(monitor_id=monitor.id)
+        assert len(checkins) == 0
+
+    @mock.patch("sentry.quotas.backend.assign_monitor_seat")
+    @mock.patch("sentry.quotas.backend.check_accept_monitor_checkin")
+    def test_monitor_accept_upsert_with_seat(
+        self,
+        check_accept_monitor_checkin,
+        assign_monitor_seat,
+    ):
+        """
+        Validates that a monitor can be upserted and processes a full check-in
+        when the PermitCheckInStatus is ACCEPTED_FOR_UPSERT and a seat is
+        allocated with a Outcome.ACCEPTED.
+        """
+        check_accept_monitor_checkin.return_value = PermitCheckInStatus.ACCEPTED_FOR_UPSERT
+        assign_monitor_seat.return_value = Outcome.ACCEPTED
+
+        self.send_checkin(
+            "my-monitor",
+            monitor_config={"schedule": {"type": "crontab", "value": "13 * * * *"}},
+            environment="my-environment",
+        )
+
+        checkin = MonitorCheckIn.objects.get(guid=self.guid)
+        assert checkin.status == CheckInStatus.OK
+
+        monitor = Monitor.objects.get(slug="my-monitor")
+        assert monitor is not None
+
+        check_accept_monitor_checkin.assert_called_with(self.project.id, monitor.slug)
+        assign_monitor_seat.assert_called_with(monitor)
+
+    @mock.patch("sentry.quotas.backend.assign_monitor_seat")
+    @mock.patch("sentry.quotas.backend.check_accept_monitor_checkin")
+    def test_monitor_accept_upsert_no_seat(
+        self,
+        check_accept_monitor_checkin,
+        assign_monitor_seat,
+    ):
+        """
+        Validates that a monitor can be upserted but have the check-in dropped
+        when the PermitCheckInStatus is ACCEPTED_FOR_UPSERT and a seat is
+        unable to be allocated with a Outcome.RATE_LIMITED
+        """
+        check_accept_monitor_checkin.return_value = PermitCheckInStatus.ACCEPTED_FOR_UPSERT
+        assign_monitor_seat.return_value = Outcome.RATE_LIMITED
+
+        self.send_checkin(
+            "my-monitor",
+            monitor_config={"schedule": {"type": "crontab", "value": "13 * * * *"}},
+            environment="my-environment",
+        )
+
+        # Check-in was not produced as we could not assign a monitor seat
+        assert not MonitorCheckIn.objects.filter(guid=self.guid).exists()
+
+        # Monitor was created, but is disabled
+        monitor = Monitor.objects.get(slug="my-monitor")
+        assert monitor is not None
+        assert monitor.status == ObjectStatus.DISABLED
+
+        check_accept_monitor_checkin.assert_called_with(self.project.id, monitor.slug)
+        assign_monitor_seat.assert_called_with(monitor)