chore(hybrid-cloud): Adds outbox maximum shard depth metric (#63551)

Author: GabeVillalobos

src/sentry/models/outbox.py

@@ -25,7 +25,7 @@
 import sentry_sdk
 from django import db
 from django.db import OperationalError, connections, models, router, transaction
-from django.db.models import Max, Min
+from django.db.models import Count, Max, Min
 from django.db.transaction import Atomic
 from django.dispatch import Signal
 from django.http import HttpRequest
@@ -675,6 +675,41 @@ def drain_shard(
                 if _test_processing_barrier:
                     _test_processing_barrier.wait()
 
+    @classmethod
+    def get_shard_depths_descending(cls, limit: Optional[int] = 10) -> list[dict[str, int | str]]:
+        """
+        Queries all outbox shards for their total depth, aggregated by their
+        sharding columns as specified by the outbox class implementation.
+
+        :param limit: Limits the query to the top N rows with the greatest shard
+        depth. If limit is None, the entire set of rows will be returned.
+        :return: A list of dictionaries, containing shard depths and shard
+        relevant column values.
+        """
+        if limit is not None:
+            assert limit > 0, "Limit must be a positive integer if specified"
+
+        base_depth_query = (
+            cls.objects.values(*cls.sharding_columns).annotate(depth=Count("*")).order_by("-depth")
+        )
+
+        if limit is not None:
+            base_depth_query = base_depth_query[0:limit]
+
+        aggregated_shard_information = list()
+        for shard_row in base_depth_query:
+            shard_information = {
+                shard_column: shard_row[shard_column] for shard_column in cls.sharding_columns
+            }
+            shard_information["depth"] = shard_row["depth"]
+            aggregated_shard_information.append(shard_information)
+
+        return aggregated_shard_information
+
+    @classmethod
+    def get_total_outbox_count(cls) -> int:
+        return cls.objects.count()
+
 
 # Outboxes bound from region silo -> control silo
 class RegionOutboxBase(OutboxBase):

src/sentry/tasks/deliver_from_outbox.py

@@ -77,10 +77,11 @@ def schedule_batch(
             scheduled_count += hi - lo + 1
             batch_size = math.ceil((hi - lo + 1) / concurrency)
 
+            metrics_tags = dict(silo_mode=silo_mode.name, outbox_name=outbox_name)
             metrics.gauge(
                 "deliver_from_outbox.queued_batch_size",
                 value=batch_size,
-                tags=dict(silo_mode=silo_mode.name),
+                tags=metrics_tags,
                 sample_rate=1.0,
             )
 
@@ -92,8 +93,28 @@ def schedule_batch(
                     outbox_identifier_low=lo + i * batch_size,
                     outbox_identifier_hi=lo + (i + 1) * batch_size,
                 )
+
+            deepest_shard_information = outbox_model.get_shard_depths_descending(limit=1)
+            max_shard_depth = (
+                deepest_shard_information[0]["depth"] if deepest_shard_information else 0
+            )
+            metrics.gauge(
+                "deliver_from_outbox.maximum_shard_depth",
+                value=max_shard_depth,
+                tags=metrics_tags,
+                sample_rate=1.0,
+            )
+
+            outbox_count = outbox_model.get_total_outbox_count()
+            metrics.gauge(
+                "deliver_from_outbox.total_outbox_count",
+                value=outbox_count,
+                tags=metrics_tags,
+                sample_rate=1.0,
+            )
         if process_outbox_backfills:
             backfill_outboxes_for(silo_mode, scheduled_count)
+
     except Exception:
         sentry_sdk.capture_exception()
         raise

tests/sentry/models/test_outbox.py

@@ -649,3 +649,64 @@ def test_bulk_operations(self):
 
         assert RegionOutbox.objects.count() == 0
         assert OrganizationMemberTeamReplica.objects.count() == 1
+
+
+class OutboxAggregationTest(TestCase):
+    def setUp(self):
+        shard_counts = {1: (4, "eu"), 2: (7, "us"), 3: (1, "us")}
+        with outbox_runner():
+            pass
+
+        for shard_id, (shard_count, region_name) in shard_counts.items():
+            for i in range(shard_count):
+                ControlOutbox(
+                    region_name=region_name,
+                    shard_scope=OutboxScope.WEBHOOK_SCOPE,
+                    shard_identifier=shard_id,
+                    category=OutboxCategory.WEBHOOK_PROXY,
+                    object_identifier=shard_id * 10000 + i,
+                    payload='{"foo": "bar"}',
+                ).save()
+
+    def test_calculate_sharding_depths(self):
+        shard_depths = ControlOutbox.get_shard_depths_descending()
+
+        assert shard_depths == [
+            dict(
+                shard_identifier=2,
+                region_name="us",
+                shard_scope=OutboxScope.WEBHOOK_SCOPE.value,
+                depth=7,
+            ),
+            dict(
+                shard_identifier=1,
+                region_name="eu",
+                shard_scope=OutboxScope.WEBHOOK_SCOPE.value,
+                depth=4,
+            ),
+            dict(
+                shard_identifier=3,
+                region_name="us",
+                shard_scope=OutboxScope.WEBHOOK_SCOPE.value,
+                depth=1,
+            ),
+        ]
+
+        # Test limiting the query to a single entry
+        shard_depths = ControlOutbox.get_shard_depths_descending(limit=1)
+        assert shard_depths == [
+            dict(
+                shard_identifier=2,
+                region_name="us",
+                shard_scope=OutboxScope.WEBHOOK_SCOPE.value,
+                depth=7,
+            )
+        ]
+
+    def test_calculate_sharding_depths_empty(self):
+        ControlOutbox.objects.all().delete()
+        assert ControlOutbox.objects.count() == 0
+        assert ControlOutbox.get_shard_depths_descending() == []
+
+    def test_total_count(self):
+        assert ControlOutbox.get_total_outbox_count() == 7 + 4 + 1