feat(aci): Enforce quota limits when creating metric detectors

src/sentry/features/temporary.py

@@ -529,6 +529,8 @@ def register_temporary_features(manager: FeatureManager):
     manager.add("organizations:workflow-engine-rule-serializers", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
     # Enable single processing of metric issues
     manager.add("organizations:workflow-engine-single-process-metric-issues", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
+    # Enable metric detector limits by plan type
+    manager.add("organizations:workflow-engine-metric-detector-limit", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Enable EventUniqueUserFrequencyConditionWithConditions special alert condition
     manager.add("organizations:event-unique-user-frequency-condition-with-conditions", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Use spans instead of transactions for dynamic sampling calculations. This will become the new default.

src/sentry/incidents/endpoints/organization_alert_rule_index.py

@@ -12,7 +12,7 @@
 from rest_framework.request import Request
 from rest_framework.response import Response
 
-from sentry import features, options
+from sentry import features, options, quotas
 from sentry.api.api_owners import ApiOwner
 from sentry.api.api_publish_status import ApiPublishStatus
 from sentry.api.base import Endpoint, region_silo_endpoint
@@ -562,6 +562,19 @@ def check_can_create_alert(self, request: Request, organization: Organization) -
         permission, then we must verify that the user is a team admin with "alerts:write" access to the project(s)
         in their request.
         """
+        if features.has(
+            "organizations:workflow-engine-metric-detector-limit", organization, actor=request.user
+        ):
+            alert_limit = quotas.backend.get_metric_detector_limit(organization.id)
+            alert_count = AlertRule.objects.fetch_for_organization(
+                organization=organization
+            ).count()
+
+            if alert_limit >= 0 and alert_count >= alert_limit:
+                raise ValidationError(
+                    f"You may not exceed {alert_limit} metric alerts on your current plan."
+                )
+
         # if the requesting user has any of these org-level permissions, then they can create an alert
         if (
             request.access.has_scope("alerts:write")

src/sentry/incidents/metric_issue_detector.py

@@ -3,6 +3,8 @@
 
 from rest_framework import serializers
 
+from sentry import features, quotas
+from sentry.constants import ObjectStatus
 from sentry.incidents.logic import enable_disable_subscriptions
 from sentry.snuba.models import QuerySubscription, SnubaQuery, SnubaQueryEventType
 from sentry.snuba.snuba_query_validator import SnubaQueryValidator
@@ -88,6 +90,32 @@ def validate(self, attrs):
 
         return attrs
 
+    def enforce_quota_on_create(self) -> None:
+        """
+        Enforce quota limits for metric detector creation.
+        Raises ValidationError if quota limits are exceeded.
+        """
+        organization = self.context.get("organization")
+        request = self.context.get("request")
+
+        if not organization or not request:
+            raise serializers.ValidationError()
+
+        if features.has(
+            "organizations:workflow-engine-metric-detector-limit", organization, actor=request.user
+        ):
+            detector_limit = quotas.backend.get_metric_detector_limit(organization.id)
+            detector_count = Detector.objects.filter(
+                project__organization=organization,
+                type="metric_issue",  # Avoided circular import. TODO: move magic strings to constant file
+                status=ObjectStatus.ACTIVE,
+            ).count()
+
+            if detector_limit >= 0 and detector_count >= detector_limit:
+                raise serializers.ValidationError(
+                    f"You may not exceed {detector_limit} metric detectors on your current plan."
+                )
+
     def update_data_source(self, instance: Detector, data_source: SnubaQueryDataSourceType):
         try:
             source_instance = DataSource.objects.get(detector=instance)

src/sentry/quotas/base.py

@@ -748,3 +748,9 @@ def get_dashboard_limit(self, org_id: int) -> int:
         Returns the maximum number of dashboards allowed for the organization's plan type.
         """
         return -1
+
+    def get_metric_detector_limit(self, org_id: int) -> int:
+        """
+        Returns the maximum number of detectors allowed for the organization's plan type.
+        """
+        return -1

src/sentry/workflow_engine/endpoints/organization_detector_index.py

@@ -314,7 +314,7 @@ def post(self, request: Request, organization: Organization) -> Response:
         if not detector_type:
             raise ValidationError({"type": ["This field is required."]})
 
-        # restrict creating metric issue detectors by plan type
+        # Restrict creating metric issue detectors by plan type
         if detector_type == MetricIssue.slug and not features.has(
             "organizations:incidents", organization, actor=request.user
         ):

src/sentry/workflow_engine/endpoints/validators/base/detector.py

@@ -64,6 +64,15 @@ def data_source(self) -> BaseDataSourceValidator:
     def data_conditions(self) -> BaseDataConditionValidator:
         raise NotImplementedError
 
+    def enforce_quota_on_create(self) -> None:
+        """
+        Enforce quota limits for detector creation.
+        Raise ValidationError if quota limits are exceeded.
+
+        Only Metric Issues Detector has quota limits.
+        """
+        pass
+
     def update(self, instance: Detector, validated_data: dict[str, Any]):
         with transaction.atomic(router.db_for_write(Detector)):
             if "name" in validated_data:
@@ -110,6 +119,10 @@ def update(self, instance: Detector, validated_data: dict[str, Any]):
         return instance
 
     def create(self, validated_data):
+        # If quotas are exceeded, we will prevent creation of new detectors.
+        # Do not disable or prevent the users from updating existing detectors.
+        self.enforce_quota_on_create()
+
         with transaction.atomic(router.db_for_write(Detector)):
             condition_group = DataConditionGroup.objects.create(
                 logic_type=DataConditionGroup.Type.ANY,

tests/sentry/incidents/endpoints/test_organization_alert_rule_index.py

@@ -30,6 +30,7 @@
     AlertRuleDetectionType,
     AlertRuleSeasonality,
     AlertRuleSensitivity,
+    AlertRuleStatus,
     AlertRuleThresholdType,
     AlertRuleTrigger,
     AlertRuleTriggerAction,
@@ -1640,6 +1641,58 @@ def test_performance_alert(self, record_analytics: MagicMock) -> None:
             resp = self.get_response(self.organization.slug, **valid_alert_rule)
             assert resp.status_code == 201
 
+    @with_feature("organizations:incidents")
+    @with_feature("organizations:workflow-engine-metric-detector-limit")
+    @patch("sentry.quotas.backend.get_metric_detector_limit")
+    def test_metric_alert_limit(self, mock_get_limit: MagicMock) -> None:
+        # Set limit to 2 alert rules
+        mock_get_limit.return_value = 2
+
+        # Create 2 existing metric alert rules (1 active, 1 to be deleted)
+        alert_rule = self.create_alert_rule(organization=self.organization)
+        alert_rule.status = AlertRuleStatus.PENDING.value
+        alert_rule.save()
+
+        alert_rule = self.create_alert_rule(organization=self.organization)
+        alert_rule.status = AlertRuleStatus.SNAPSHOT.value
+        alert_rule.save()
+
+        # Create another alert rule, it should succeed
+        data = deepcopy(self.alert_rule_dict)
+        with outbox_runner():
+            resp = self.get_success_response(
+                self.organization.slug,
+                status_code=201,
+                **data,
+            )
+        alert_rule = AlertRule.objects.get(id=resp.data["id"])
+        assert alert_rule.name == "JustAValidTestRule"
+
+        # Create another alert rule, it should fail
+        data = deepcopy(self.alert_rule_dict)
+        resp = self.get_error_response(
+            self.organization.slug,
+            status_code=400,
+            **data,
+        )
+
+    @with_feature("organizations:incidents")
+    @with_feature("organizations:workflow-engine-metric-detector-limit")
+    def test_metric_alert_limit_unlimited_plan(self) -> None:
+        # Create many alert rules
+        for _ in range(5):
+            self.create_alert_rule(organization=self.organization)
+
+        # Creating another alert rule, it should succeed
+        with outbox_runner():
+            resp = self.get_success_response(
+                self.organization.slug,
+                status_code=201,
+                **self.alert_rule_dict,
+            )
+        alert_rule = AlertRule.objects.get(id=resp.data["id"])
+        assert alert_rule.name == "JustAValidTestRule"
+
 
 @freeze_time()
 class AlertRuleCreateEndpointTestCrashRateAlert(AlertRuleIndexBase):

tests/sentry/incidents/endpoints/validators/test_validators.py

@@ -1,8 +1,9 @@
 from unittest import mock
 
-from rest_framework.exceptions import ErrorDetail
+from rest_framework.exceptions import ErrorDetail, ValidationError
 
 from sentry import audit_log
+from sentry.constants import ObjectStatus
 from sentry.incidents.grouptype import MetricIssue
 from sentry.incidents.metric_issue_detector import (
     MetricIssueComparisonConditionValidator,
@@ -23,6 +24,7 @@
     SnubaQuery,
     SnubaQueryEventType,
 )
+from sentry.testutils.helpers.features import with_feature
 from sentry.workflow_engine.models import DataCondition, DataConditionGroup, DataSource, Detector
 from sentry.workflow_engine.models.data_condition import Condition
 from sentry.workflow_engine.registry import data_source_type_registry
@@ -358,3 +360,49 @@ def test_too_many_conditions(self) -> None:
         assert validator.errors.get("nonFieldErrors") == [
             ErrorDetail(string="Too many conditions", code="invalid")
         ]
+
+    @mock.patch("sentry.quotas.backend.get_metric_detector_limit")
+    def test_enforce_quota_feature_disabled(self, mock_get_limit: mock.MagicMock) -> None:
+        mock_get_limit.return_value = 0
+        validator = MetricIssueDetectorValidator(data=self.valid_data, context=self.context)
+
+        assert validator.is_valid()
+        assert validator.save()
+        mock_get_limit.assert_not_called()
+
+    @mock.patch("sentry.quotas.backend.get_metric_detector_limit")
+    @with_feature("organizations:workflow-engine-metric-detector-limit")
+    def test_enforce_quota_within_limit(self, mock_get_limit: mock.MagicMock) -> None:
+        mock_get_limit.return_value = 1
+
+        # Create a not-metric detector
+        self.create_detector(
+            project_id=self.project.id,
+            name="Error Detector",
+            status=ObjectStatus.ACTIVE,
+        )
+        # Create 3 inactive detectors
+        for status in [
+            ObjectStatus.DISABLED,
+            ObjectStatus.PENDING_DELETION,
+            ObjectStatus.DELETION_IN_PROGRESS,
+        ]:
+            self.create_detector(
+                project_id=self.project.id,
+                name=f"Inactive Detector {status}",
+                type=MetricIssue.slug,
+                status=status,
+            )
+
+        validator = MetricIssueDetectorValidator(data=self.valid_data, context=self.context)
+        assert validator.is_valid()
+        assert validator.save()
+        mock_get_limit.assert_called_once_with(self.project.organization.id)
+
+        validator = MetricIssueDetectorValidator(data=self.valid_data, context=self.context)
+        validator.is_valid()
+        with self.assertRaisesMessage(
+            ValidationError,
+            expected_message="You may not exceed 1 metric detectors on your current plan.",
+        ):
+            validator.save()

tests/sentry/workflow_engine/endpoints/test_organization_detector_index.py

@@ -956,6 +956,115 @@ def test_owner_not_in_organization(self) -> None:
         )
         assert "owner" in response.data
 
+    @with_feature("organizations:workflow-engine-metric-detector-limit")
+    @mock.patch("sentry.quotas.backend.get_metric_detector_limit")
+    def test_metric_detector_limit(self, mock_get_limit: mock.MagicMock) -> None:
+        # Set limit to 2 detectors
+        mock_get_limit.return_value = 2
+
+        # Create 2 metric detectors (1 active, 1 to be deleted)
+        self.create_detector(
+            project_id=self.project.id,
+            name="Existing Detector 1",
+            type=MetricIssue.slug,
+            status=ObjectStatus.ACTIVE,
+        )
+        self.create_detector(
+            project_id=self.project.id,
+            name="Existing Detector 2",
+            type=MetricIssue.slug,
+            status=ObjectStatus.PENDING_DELETION,
+        )
+
+        # Create another metric detector, it should succeed
+        with self.tasks():
+            response = self.get_success_response(
+                self.organization.slug,
+                **self.valid_data,
+                status_code=201,
+            )
+        detector = Detector.objects.get(id=response.data["id"])
+        assert detector.name == "Test Detector"
+
+        # Create another metric detector, it should fail
+        response = self.get_error_response(
+            self.organization.slug,
+            **self.valid_data,
+            status_code=400,
+        )
+        assert response.status_code == 400
+
+    @with_feature("organizations:workflow-engine-metric-detector-limit")
+    @mock.patch("sentry.quotas.backend.get_metric_detector_limit")
+    def test_metric_detector_limit_unlimited_plan(self, mock_get_limit: mock.MagicMock) -> None:
+        # Set limit to -1 (unlimited)
+        mock_get_limit.return_value = -1
+
+        # Create many metric detectors
+        for i in range(5):
+            self.create_detector(
+                project_id=self.project.id,
+                name=f"Existing Detector {i+1}",
+                type=MetricIssue.slug,
+                status=ObjectStatus.ACTIVE,
+            )
+
+        # Create another detector, it should succeed
+        with self.tasks():
+            response = self.get_success_response(
+                self.organization.slug,
+                **self.valid_data,
+                status_code=201,
+            )
+        mock_get_limit.assert_called_once_with(self.organization.id)
+        detector = Detector.objects.get(id=response.data["id"])
+        assert detector.name == "Test Detector"
+
+    @with_feature("organizations:workflow-engine-metric-detector-limit")
+    @mock.patch("sentry.quotas.backend.get_metric_detector_limit")
+    def test_metric_detector_limit_only_applies_to_metric_detectors(
+        self, mock_get_limit: mock.MagicMock
+    ) -> None:
+        # Set limit to 1 metric detector
+        mock_get_limit.return_value = 1
+
+        # Create a not-metric detector
+        self.create_detector(
+            project_id=self.project.id,
+            name="Error Detector",
+            type=ErrorGroupType.slug,
+            status=ObjectStatus.ACTIVE,
+        )
+
+        # Create 1 metric detector, it should succeed
+        response = self.get_success_response(
+            self.organization.slug,
+            **self.valid_data,
+            status_code=201,
+        )
+        detector = Detector.objects.get(id=response.data["id"])
+        assert detector.name == "Test Detector"
+
+        # Create another metric detector, it should fail
+        response = self.get_error_response(
+            self.organization.slug,
+            **self.valid_data,
+            status_code=400,
+        )
+        assert response.status_code == 400
+
+        # Create another not-metric detector, it should succeed
+        with self.tasks():
+            response = self.get_success_response(
+                self.organization.slug,
+                projectId=self.project.id,
+                name="Error Detector",
+                type=ErrorGroupType.slug,
+                status_code=201,
+            )
+        detector = Detector.objects.get(id=response.data["id"])
+        assert detector.type == ErrorGroupType.slug
+
 
 @region_silo_test
 @with_feature("organizations:incidents")