diff --git a/src/sentry/hybridcloud/tasks/webhook_backlog_metrics.py b/src/sentry/hybridcloud/tasks/webhook_backlog_metrics.py index 62b86507f6e2..5af231cb1d8a 100644 --- a/src/sentry/hybridcloud/tasks/webhook_backlog_metrics.py +++ b/src/sentry/hybridcloud/tasks/webhook_backlog_metrics.py @@ -1,7 +1,8 @@ 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 @@ -28,6 +29,12 @@ bounds it, so it gets an explicit bound. """ +DEPTH_QUANTILES = (50, 90, 99) +""" +`max_depth` reads the same whether one mailbox is stuck or a provider's whole backlog +has shifted deeper. One is a mailbox to go look at, the other a capacity problem. +""" + MAILBOX_DEPTH_QUERY_TIMEOUT = datetime.timedelta(seconds=30) """ Ceiling on the per-mailbox aggregate below. @@ -39,6 +46,17 @@ """ +def _nearest_rank(sorted_depths: Sequence[int], percentile: int) -> int: + """ + Depth of the mailbox at `percentile` of an ascending `sorted_depths`. + + Nearest-rank, so every value returned is a depth some mailbox actually has. The + clamp stops a low percentile from indexing -1 and returning the deepest instead. + """ + 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]: """ @@ -146,6 +164,9 @@ 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. + + `depth_quantile` gives the shape behind `max_depth`. The depths are already in + memory for the aggregates above, so it costs one sort per provider. """ replica = WebhookPayload.objects.using_replica() mailboxes = replica.values("provider", "mailbox_name").annotate( @@ -162,18 +183,16 @@ def record_mailbox_depth_metrics() -> None: 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] = {} + depths: dict[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) + depths[provider].append(depth) for (provider, event_type), pending_count in pending.items(): metrics.gauge( @@ -183,17 +202,25 @@ def record_mailbox_depth_metrics() -> None: sample_rate=1.0, ) - for provider, active_count in mailbox_count.items(): + for provider, mailbox_depths in depths.items(): + mailbox_depths.sort() tags = {"provider": provider} + for percentile in DEPTH_QUANTILES: + metrics.gauge( + "hybridcloud.webhookpayload.mailbox.depth_quantile", + _nearest_rank(mailbox_depths, percentile), + tags={**tags, "quantile": f"p{percentile}"}, + sample_rate=1.0, + ) metrics.gauge( "hybridcloud.webhookpayload.mailbox.active_count", - active_count, + len(mailbox_depths), tags=tags, sample_rate=1.0, ) metrics.gauge( "hybridcloud.webhookpayload.mailbox.max_depth", - max_depth[provider], + mailbox_depths[-1], tags=tags, sample_rate=1.0, ) diff --git a/tests/sentry/hybridcloud/tasks/test_webhook_backlog_metrics.py b/tests/sentry/hybridcloud/tasks/test_webhook_backlog_metrics.py index 51cc7d4c6c8f..f5409939e257 100644 --- a/tests/sentry/hybridcloud/tasks/test_webhook_backlog_metrics.py +++ b/tests/sentry/hybridcloud/tasks/test_webhook_backlog_metrics.py @@ -23,6 +23,7 @@ MAILBOX_ACTIVE_METRIC = "hybridcloud.webhookpayload.mailbox.active_count" MAILBOX_MAX_DEPTH_METRIC = "hybridcloud.webhookpayload.mailbox.max_depth" MAILBOX_AGE_METRIC = "hybridcloud.webhookpayload.mailbox.oldest_pending_age_seconds" +MAILBOX_DEPTH_QUANTILE_METRIC = "hybridcloud.webhookpayload.mailbox.depth_quantile" def create_payloads(num: int, mailbox: str, provider: str | None = None) -> None: @@ -39,6 +40,15 @@ def gauge_calls(mock_metrics: MagicMock, key: str) -> list[tuple[float, dict[str ] +def depth_quantiles(mock_metrics: MagicMock, provider: str) -> dict[str, float]: + """{quantile: depth} reported for `provider`.""" + return { + tags["quantile"]: value + for value, tags in gauge_calls(mock_metrics, MAILBOX_DEPTH_QUANTILE_METRIC) + if tags["provider"] == provider + } + + @control_silo_test class WebhookBacklogMetricsTest(TestCase): @patch("sentry.hybridcloud.tasks.webhook_backlog_metrics.metrics") @@ -312,6 +322,40 @@ def test_only_pending_count_is_tagged_by_event_type(self, mock_metrics: MagicMoc assert [tags for _, tags in gauge_calls(mock_metrics, metric)] == [ {"provider": "github"} ], metric + assert all( + "event_type" not in tags + for _, tags in gauge_calls(mock_metrics, MAILBOX_DEPTH_QUANTILE_METRIC) + ) + + @patch("sentry.hybridcloud.tasks.webhook_backlog_metrics.metrics") + def test_depth_quantiles_describe_the_distribution(self, mock_metrics: MagicMock) -> None: + for depth, mailbox in enumerate(("a", "b", "c", "d"), start=1): + create_payloads(depth, f"github:{mailbox}:push", provider="github") + + # Nearest-rank over [1, 2, 3, 4]. + record_mailbox_depth_metrics() + + assert depth_quantiles(mock_metrics, "github") == {"p50": 2, "p90": 4, "p99": 4} + + @patch("sentry.hybridcloud.tasks.webhook_backlog_metrics.metrics") + def test_depth_quantiles_hold_for_a_single_mailbox(self, mock_metrics: MagicMock) -> None: + create_payloads(5, "github:123:push", provider="github") + + record_mailbox_depth_metrics() + + # Covers the index clamp; a wrap to -1 is only visible in the test above. + assert depth_quantiles(mock_metrics, "github") == {"p50": 5, "p90": 5, "p99": 5} + + @patch("sentry.hybridcloud.tasks.webhook_backlog_metrics.metrics") + def test_depth_quantiles_are_per_provider(self, mock_metrics: MagicMock) -> None: + create_payloads(1, "github:1:push", provider="github") + create_payloads(9, "github:2:push", provider="github") + create_payloads(4, "gitlab:3", provider="gitlab") + + record_mailbox_depth_metrics() + + assert depth_quantiles(mock_metrics, "github") == {"p50": 1, "p90": 9, "p99": 9} + assert depth_quantiles(mock_metrics, "gitlab") == {"p50": 4, "p90": 4, "p99": 4} def test_aggregate_runs_under_a_statement_timeout(self) -> None: create_payloads(1, "github:123", provider="github")