ref(uptime): Simplify consumer with one-to-one proj_sub <-> sub (take 2) (#90261)

This reverts commit f08b664 with fixes
to handle when there is no project_subscription for a
uptime_subscription

Author: evanpurkhiser

src/sentry/uptime/consumers/results_consumer.py

@@ -33,13 +33,14 @@
     UptimeSubscription,
     UptimeSubscriptionRegion,
     get_detector,
-    get_project_subscriptions_for_uptime_subscription,
+    get_project_subscription_for_uptime_subscription,
     get_top_hosting_provider_names,
     load_regions_for_uptime_subscription,
 )
 from sentry.uptime.subscriptions.subscriptions import (
     check_and_update_regions,
     delete_uptime_detector,
+    remove_uptime_subscription_if_unused,
     update_project_uptime_subscription,
 )
 from sentry.uptime.subscriptions.tasks import (
@@ -293,29 +294,17 @@ def handle_result(self, subscription: UptimeSubscription | None, result: CheckRe
 
         try_check_and_update_regions(subscription, result, subscription_regions)
 
-        project_subscriptions = get_project_subscriptions_for_uptime_subscription(subscription.id)
+        project_subscription = get_project_subscription_for_uptime_subscription(subscription.id)
 
-        cluster = _get_cluster()
-        last_updates: list[str | None] = cluster.mget(
-            build_last_update_key(sub) for sub in project_subscriptions
-        )
+        # Nothing to do if there's an orphaned project subscription
+        if not project_subscription:
+            remove_uptime_subscription_if_unused(subscription)
+            return
 
-        for last_update_raw, project_subscription in zip(last_updates, project_subscriptions):
-            last_update_ms = 0 if last_update_raw is None else int(last_update_raw)
-            self.handle_result_for_project(
-                project_subscription,
-                result,
-                last_update_ms,
-                metric_tags.copy(),
-            )
+        cluster = _get_cluster()
+        last_update_raw: str | None = cluster.get(build_last_update_key(project_subscription))
+        last_update_ms = 0 if last_update_raw is None else int(last_update_raw)
 
-    def handle_result_for_project(
-        self,
-        project_subscription: ProjectUptimeSubscription,
-        result: CheckResult,
-        last_update_ms: int,
-        metric_tags: dict[str, str],
-    ):
         if features.has(
             "organizations:uptime-detailed-logging", project_subscription.project.organization
         ):

src/sentry/uptime/models.py

@@ -266,14 +266,15 @@ def get_top_hosting_provider_names(limit: int) -> set[str]:
     recalculate=False,
     cache_ttl=timedelta(hours=4),
 )
-def get_project_subscriptions_for_uptime_subscription(
+def get_project_subscription_for_uptime_subscription(
     uptime_subscription_id: int,
-) -> list[ProjectUptimeSubscription]:
-    return list(
-        ProjectUptimeSubscription.objects.filter(
-            uptime_subscription_id=uptime_subscription_id
-        ).select_related("project", "project__organization")
-    )
+) -> ProjectUptimeSubscription | None:
+    try:
+        return ProjectUptimeSubscription.objects.select_related(
+            "project", "project__organization"
+        ).get(uptime_subscription_id=uptime_subscription_id)
+    except ProjectUptimeSubscription.DoesNotExist:
+        return None
 
 
 @cache_func_for_models(

tests/sentry/uptime/consumers/test_results_consumer.py

@@ -22,7 +22,7 @@
 
 from sentry.conf.types import kafka_definition
 from sentry.conf.types.uptime import UptimeRegionConfig
-from sentry.constants import DataCategory, ObjectStatus
+from sentry.constants import DataCategory
 from sentry.models.group import Group, GroupStatus
 from sentry.testutils.abstract import Abstract
 from sentry.testutils.helpers.datetime import freeze_time
@@ -157,6 +157,25 @@ def test(self):
         assert self.project_subscription.uptime_status == UptimeStatus.FAILED
         assert self.project_subscription.uptime_subscription.uptime_status == UptimeStatus.FAILED
 
+    def test_does_nothing_when_missing_project_subscription(self):
+        self.project_subscription.delete()
+
+        result = self.create_uptime_result(
+            self.subscription.subscription_id,
+            scheduled_check_time=datetime.now() - timedelta(minutes=5),
+        )
+        with (
+            self.feature(["organizations:uptime", "organizations:uptime-create-issues"]),
+            mock.patch("sentry.remote_subscriptions.consumers.result_consumer.logger") as logger,
+            mock.patch(
+                "sentry.uptime.consumers.results_consumer.remove_uptime_subscription_if_unused"
+            ) as mock_remove_uptime_subscription_if_unused,
+        ):
+            # Does not produce an error
+            self.send_result(result)
+            assert not logger.exception.called
+            mock_remove_uptime_subscription_if_unused.assert_called_with(self.subscription)
+
     def test_restricted_host_provider_id(self):
         """
         Test that we do NOT create an issue when the host provider identifier
@@ -446,32 +465,6 @@ def test_no_subscription(self):
                 "default", UptimeSubscription(subscription_id=subscription_id), "delete", None
             )
 
-    def test_multiple_project_subscriptions_with_disabled(self):
-        """
-        Tests that we do not process results for disabled project subscriptions
-        """
-        # Second disabled project subscription
-        self.create_project_uptime_subscription(
-            uptime_subscription=self.subscription,
-            project=self.create_project(),
-            status=ObjectStatus.DISABLED,
-        )
-        result = self.create_uptime_result(self.subscription.subscription_id)
-
-        with (
-            mock.patch("sentry.uptime.consumers.results_consumer.metrics") as metrics,
-            self.feature(["organizations:uptime", "organizations:uptime-create-issues"]),
-        ):
-            self.send_result(result)
-            # We only process a single project result, the other is dropped,
-            # there should be only one handle_result_for_project metric call
-            handle_result_calls = [
-                c
-                for c in metrics.incr.mock_calls
-                if c[1][0] == "uptime.result_processor.handle_result_for_project"
-            ]
-            assert len(handle_result_calls) == 1
-
     def test_organization_feature_disabled(self):
         """
         Tests that we do not process results for disabled project subscriptions
@@ -1063,6 +1056,7 @@ def test_check_and_update_regions(self):
             subscription_id=uuid.UUID(int=5).hex,
             region_slugs=["region1"],
         )
+        self.create_project_uptime_subscription(uptime_subscription=sub)
         self.run_check_and_update_region_test(
             sub,
             ["region1", "region2"],
@@ -1093,6 +1087,7 @@ def test_check_and_update_regions_active_shadow(self):
             subscription_id=uuid.UUID(int=5).hex,
             region_slugs=["region1", "region2"],
         )
+        self.create_project_uptime_subscription(uptime_subscription=sub)
         self.run_check_and_update_region_test(
             sub,
             ["region1", "region2"],
@@ -1119,6 +1114,7 @@ def test_check_and_update_regions_larger_interval(self):
             region_slugs=["region1"],
             interval_seconds=UptimeSubscription.IntervalSeconds.ONE_HOUR,
         )
+        self.create_project_uptime_subscription(uptime_subscription=hour_sub)
         self.run_check_and_update_region_test(
             hour_sub,
             ["region1", "region2"],
@@ -1140,6 +1136,7 @@ def test_check_and_update_regions_larger_interval(self):
             region_slugs=["region1"],
             interval_seconds=UptimeSubscription.IntervalSeconds.FIVE_MINUTES,
         )
+        self.create_project_uptime_subscription(uptime_subscription=five_min_sub)
         self.run_check_and_update_region_test(
             five_min_sub,
             ["region1", "region2"],
@@ -1188,6 +1185,7 @@ def test_check_and_update_regions_larger_interval(self):
             region_slugs=["region1"],
             interval_seconds=UptimeSubscription.IntervalSeconds.FIVE_MINUTES,
         )
+        self.create_project_uptime_subscription(uptime_subscription=five_min_sub)
         self.run_check_and_update_region_test(
             five_min_sub,
             ["region1", "region2"],
@@ -1206,8 +1204,10 @@ def test_check_and_update_regions_larger_interval(self):
 
     def test_check_and_update_regions_removes_disabled(self):
         sub = self.create_uptime_subscription(
-            subscription_id=uuid.UUID(int=5).hex, region_slugs=["region1", "region2"]
+            subscription_id=uuid.UUID(int=5).hex,
+            region_slugs=["region1", "region2"],
         )
+        self.create_project_uptime_subscription(uptime_subscription=sub)
         self.run_check_and_update_region_test(
             sub,
             ["region1", "region2"],
@@ -1259,6 +1259,7 @@ def test_parallel(self) -> None:
             subscription_2 = self.create_uptime_subscription(
                 subscription_id=uuid.uuid4().hex, interval_seconds=300, url="http://santry.io"
             )
+            self.create_project_uptime_subscription(uptime_subscription=subscription_2)
 
             result_1 = self.create_uptime_result(
                 self.subscription.subscription_id,
@@ -1310,6 +1311,7 @@ def test_parallel_grouping(self, mock_process_group) -> None:
         subscription_2 = self.create_uptime_subscription(
             subscription_id=uuid.uuid4().hex, interval_seconds=300, url="http://santry.io"
         )
+        self.create_project_uptime_subscription(uptime_subscription=subscription_2)
 
         result_1 = self.create_uptime_result(
             self.subscription.subscription_id,