feat(hybridcloud): Split backlog metrics by dispatch regime and report depth quantiles

src/sentry/hybridcloud/tasks/deliver_webhooks.py

@@ -197,7 +197,7 @@ def _claim_dispatch_active() -> bool:
     return options.get("hybridcloud.webhookpayload.claim_dispatch_rollout") > 0.0
 
 
-def _use_claim_dispatch(mailbox_name: str) -> bool:
+def use_claim_dispatch(mailbox_name: str) -> bool:
     """
     Whether this mailbox dispatches drains via batch claims instead of the
     drain-lock lease, per the claim_dispatch_rollout rate.
@@ -495,14 +495,14 @@ def _maybe_trigger_drain_lease(mailbox_name: str) -> None:
 def maybe_trigger_drain(mailbox_name: str) -> None:
     """Trigger an immediate drain if the mailbox head is due for delivery.
 
-    While claim_dispatch_rollout ramps, `_use_claim_dispatch` picks between the
+    While claim_dispatch_rollout ramps, `use_claim_dispatch` picks between the
     claim and lease trigger regimes (see the two helpers above).
 
     Falls back gracefully if the cache backend is unavailable — the scheduler handles delivery.
     """
     if not options.get("hybridcloud.webhookpayload.push_drain_trigger"):
         return
-    if _use_claim_dispatch(mailbox_name):
+    if use_claim_dispatch(mailbox_name):
         _maybe_trigger_drain_claim(mailbox_name)
     else:
         _maybe_trigger_drain_lease(mailbox_name)
@@ -567,7 +567,7 @@ def schedule_webhook_delivery() -> None:
 
     for record in scheduled_mailboxes[:BATCH_SIZE]:
         mailbox_name = record["mailbox_name"]
-        if not _use_claim_dispatch(mailbox_name):
+        if not use_claim_dispatch(mailbox_name):
             # Lease-mode mailbox (the legacy path): skip anything a push-triggered
             # drain currently holds, then claim and dispatch without the guard.
             if options.get("hybridcloud.webhookpayload.push_drain_trigger"):
@@ -637,7 +637,7 @@ def drain_mailbox(
     started a drain of its own. `None` (lease drains, tasks enqueued before this
     field deployed) keeps the legacy behavior of draining until empty.
 
-    `mailbox_name` is sent by lease-mode push triggers (see `_use_claim_dispatch`),
+    `mailbox_name` is sent by lease-mode push triggers (see `use_claim_dispatch`),
     which hand this drain ownership of the drain lock for its whole run: refreshed
     on every delivery, released on exit. Claim-mode dispatchers never send it.
 

src/sentry/hybridcloud/tasks/webhook_backlog_metrics.py

@@ -1,14 +1,16 @@
 import datetime
 import logging
+import math
 from collections import defaultdict
-from collections.abc import Generator
+from collections.abc import Generator, Sequence
 from contextlib import contextmanager
 
 from django.db import OperationalError, connections, transaction
 from django.db.models import Count, Min
 from django.utils import timezone
 
 from sentry.hybridcloud.models.webhookpayload import WebhookPayload
+from sentry.hybridcloud.tasks.deliver_webhooks import DispatchMode, use_claim_dispatch
 from sentry.integrations.github.webhook_types import CELL_PROCESSED_GITHUB_EVENTS
 from sentry.integrations.types import IntegrationProviderSlug
 from sentry.silo.base import SiloMode
@@ -29,6 +31,16 @@
 bounds it, so it gets an explicit bound.
 """
 
+DEPTH_QUANTILES = (50, 90, 99)
+"""
+Which points of the mailbox depth distribution to report.
+
+`max_depth` alone cannot distinguish one pathological mailbox from a provider whose
+whole backlog has shifted deeper, and those want different responses. Three points
+plus the existing max and mailbox count describe the shape without paying a series
+per mailbox.
+"""
+
 MAILBOX_DEPTH_QUERY_TIMEOUT = datetime.timedelta(seconds=30)
 """
 Ceiling on the per-mailbox aggregate below.
@@ -66,6 +78,18 @@ def _event_type_from_mailbox(provider: str, mailbox_name: str) -> str:
     return suffix if suffix in CELL_PROCESSED_GITHUB_EVENTS else UNKNOWN_EVENT_TYPE
 
 
+def _nearest_rank(sorted_depths: Sequence[int], percentile: int) -> int:
+    """
+    The depth of the mailbox sitting at `percentile` of an ascending `sorted_depths`.
+
+    Nearest-rank rather than interpolated: every value this returns is a depth some
+    mailbox actually has, so a reported p99 of 300 names a mailbox you can go find.
+    An interpolated 287.4 describes none of them.
+    """
+    index = math.ceil(percentile / 100 * len(sorted_depths)) - 1
+    return sorted_depths[max(index, 0)]
+
+
 @contextmanager
 def _statement_timeout(alias: str, timeout: datetime.timedelta) -> Generator[None]:
     """
@@ -173,6 +197,12 @@ def record_mailbox_depth_metrics() -> None:
     is made of and joins to `github.webhook.forwarded_event`; summing over the tag
     reproduces the provider-only value. Only that metric — the rest read fine per
     provider, and every tag value costs a series per worker that runs the task.
+
+    Every mailbox metric is tagged by the dispatch regime that would drain it, so a
+    backlog burning down under one regime and flat under the other is legible while
+    `claim_dispatch_rollout` ramps. The regime is derived per mailbox rather than
+    stored, so a rollout change re-attributes existing backlog on the next run: the
+    tag answers "which regime owns this backlog now", not "which one produced it".
     """
     replica = WebhookPayload.objects.using_replica()
     mailboxes = replica.values("provider", "mailbox_name").annotate(
@@ -188,30 +218,48 @@ def record_mailbox_depth_metrics() -> None:
         return
 
     now = timezone.now()
-    pending: dict[tuple[str, str], int] = defaultdict(int)
-    mailbox_count: dict[str, int] = defaultdict(int)
-    max_depth: dict[str, int] = defaultdict(int)
-    oldest: dict[str, datetime.datetime] = {}
+    pending: dict[tuple[str, str, str], int] = defaultdict(int)
+    mailbox_count: dict[tuple[str, str], int] = defaultdict(int)
+    max_depth: dict[tuple[str, str], int] = defaultdict(int)
+    oldest: dict[tuple[str, str], datetime.datetime] = {}
+    depths: dict[tuple[str, str], list[int]] = defaultdict(list)
     for row in rows:
         # The column is nullable, and rows predating it still drain through here.
         provider = row["provider"] or "unknown"
         depth, row_oldest = row["depth"], row["oldest"]
-        event_type = _event_type_from_mailbox(provider, row["mailbox_name"])
-        pending[(provider, event_type)] += depth
-        mailbox_count[provider] += 1
-        max_depth[provider] = max(max_depth[provider], depth)
-        oldest[provider] = min(oldest.get(provider, row_oldest), row_oldest)
+        mailbox_name = row["mailbox_name"]
+        # Reuses the dispatcher's own predicate rather than restating the hash, so the
+        # cohort this reports can never drift from the cohort that actually claims.
+        mode = (
+            DispatchMode.CLAIM if use_claim_dispatch(mailbox_name) else DispatchMode.LEASE
+        ).value
+        event_type = _event_type_from_mailbox(provider, mailbox_name)
+        pending[(provider, event_type, mode)] += depth
+        mailbox_count[(provider, mode)] += 1
+        max_depth[(provider, mode)] = max(max_depth[(provider, mode)], depth)
+        oldest[(provider, mode)] = min(oldest.get((provider, mode), row_oldest), row_oldest)
+        depths[(provider, mode)].append(depth)
 
-    for (provider, event_type), pending_count in pending.items():
+    for (provider, event_type, mode), pending_count in pending.items():
         metrics.gauge(
             "hybridcloud.webhookpayload.mailbox.pending_count",
             pending_count,
-            tags={"provider": provider, "event_type": event_type},
+            tags={"provider": provider, "event_type": event_type, "mode": mode},
             sample_rate=1.0,
         )
 
-    for provider, active_count in mailbox_count.items():
-        tags = {"provider": provider}
+    for (provider, mode), mailbox_depths in depths.items():
+        mailbox_depths.sort()
+        for percentile in DEPTH_QUANTILES:
+            metrics.gauge(
+                "hybridcloud.webhookpayload.mailbox.depth_quantile",
+                _nearest_rank(mailbox_depths, percentile),
+                tags={"provider": provider, "mode": mode, "quantile": f"p{percentile}"},
+                sample_rate=1.0,
+            )
+
+    for (provider, mode), active_count in mailbox_count.items():
+        tags = {"provider": provider, "mode": mode}
         metrics.gauge(
             "hybridcloud.webhookpayload.mailbox.active_count",
             active_count,
@@ -220,13 +268,13 @@ def record_mailbox_depth_metrics() -> None:
         )
         metrics.gauge(
             "hybridcloud.webhookpayload.mailbox.max_depth",
-            max_depth[provider],
+            max_depth[(provider, mode)],
             tags=tags,
             sample_rate=1.0,
         )
         metrics.gauge(
             "hybridcloud.webhookpayload.mailbox.oldest_pending_age_seconds",
-            (now - oldest[provider]).total_seconds(),
+            (now - oldest[(provider, mode)]).total_seconds(),
             tags=tags,
             sample_rate=1.0,
             unit="second",

tests/sentry/hybridcloud/tasks/test_deliver_webhooks.py

@@ -22,11 +22,11 @@
     Dispatcher,
     DispatchMode,
     _claim_and_dispatch,
-    _use_claim_dispatch,
     drain_mailbox,
     drain_mailbox_parallel,
     maybe_trigger_drain,
     schedule_webhook_delivery,
+    use_claim_dispatch,
 )
 from sentry.silo.client import CellSiloClient
 from sentry.testutils.cases import TestCase
@@ -2289,13 +2289,13 @@ def test_lease_trigger_respects_active_claim(self, mock_drain: MagicMock) -> Non
 
     def test_claim_rollout_buckets_by_integration(self) -> None:
         with override_options({"hybridcloud.webhookpayload.claim_dispatch_rollout": 0.0}):
-            assert _use_claim_dispatch("github:123") is False
+            assert use_claim_dispatch("github:123") is False
         with override_options({"hybridcloud.webhookpayload.claim_dispatch_rollout": 1.0}):
-            assert _use_claim_dispatch("github:123") is True
+            assert use_claim_dispatch("github:123") is True
         with override_options({"hybridcloud.webhookpayload.claim_dispatch_rollout": 0.5}):
             # Sub-mailboxes of one integration must land in the same regime as
             # their base mailbox, deterministically across calls.
-            base = _use_claim_dispatch("github:123")
-            assert _use_claim_dispatch("github:123") is base
-            assert _use_claim_dispatch("github:123:93:check_run") is base
-            assert _use_claim_dispatch("github:123:5:push") is base
+            base = use_claim_dispatch("github:123")
+            assert use_claim_dispatch("github:123") is base
+            assert use_claim_dispatch("github:123:93:check_run") is base
+            assert use_claim_dispatch("github:123:5:push") is base