From 4b2ab8c2cf597cb524d785877ee1b3bfec92d640 Mon Sep 17 00:00:00 2001 From: Ivan Dlugos Date: Thu, 13 Aug 2026 20:05:42 +0200 Subject: [PATCH 1/2] feat(hybridcloud): Tag delivery outcomes with their dispatcher and mode The claim_dispatch_rollout selects whole integrations by hash, so at a low percentage its effect on any global total is far smaller than that total's own variance. Races run ~321k/2h with a per-bucket standard deviation of ~60%; a 1% cohort moves them by ~13 per bucket, which is ~60x below one standard deviation. Reading the rollout from global totals needs ~12% of traffic over 2h, or ~3.4% over a day, before the signal clears noise. Segmenting by cohort removes that constraint: comparing the claim cohort's outcome rates against the lease cohort's is valid at any rollout percentage, because it is a rate comparison rather than a shift in a sum. `delivery` was tagged `outcome` and `provider` only. Once a drain task starts it has no idea which dispatcher enqueued it or under which regime, so neither segmentation was possible. Both drain tasks now accept `dispatcher` and `mode` and carry them onto every delivery outcome they record. This also replaces the derived push-vs-scheduler item split on the dashboard, which had to infer push volume by subtracting scheduler claims from the total. Tasks already queued when this deploys arrive without the arguments and tag `unknown` rather than omitting the keys, since a tag absent from some series breaks grouping instead of showing a gap. Requires the `dispatcher` and `mode` tags to be added to the delivery metric's Datadog tag allowlist before this deploys; it is an allowlist, so unlisted tags are dropped at intake and the change is not retroactive. No behavior change: no query, delivery decision, or claim bound is altered. --- .../hybridcloud/tasks/deliver_webhooks.py | 168 ++++++++++++++---- .../tasks/test_deliver_webhooks.py | 167 +++++++++++++++-- 2 files changed, 280 insertions(+), 55 deletions(-) diff --git a/src/sentry/hybridcloud/tasks/deliver_webhooks.py b/src/sentry/hybridcloud/tasks/deliver_webhooks.py index f9e1287922ff..fc482e97ef8a 100644 --- a/src/sentry/hybridcloud/tasks/deliver_webhooks.py +++ b/src/sentry/hybridcloud/tasks/deliver_webhooks.py @@ -1,6 +1,7 @@ import datetime import enum import logging +from collections.abc import Mapping from concurrent.futures import as_completed import orjson @@ -256,16 +257,41 @@ class DispatchOutcome(enum.StrEnum): class Dispatcher(enum.StrEnum): - """Which dispatcher enqueued a drain; a metric tag.""" + """Which dispatcher enqueued a drain; a metric tag and a drain task argument.""" PUSH = "push" SCHEDULER = "scheduler" +class DispatchMode(enum.StrEnum): + """Which dispatch regime enqueued a drain; a metric tag and a drain task argument.""" + + LEASE = "lease" + CLAIM = "claim" + + +def _dispatch_tags(dispatcher: str | None, mode: str | None) -> dict[str, str]: + """ + Attribute a drain's deliveries to the dispatcher and regime that enqueued it. + + Tagging `delivery` with `mode` is what makes a partial claim_dispatch_rollout + readable. The rollout selects whole integrations by hash, so at a low + percentage its effect on any global total is far smaller than the total's own + variance; segmenting by mode compares the claim cohort's outcome rates against + the lease cohort's instead, which stays valid at any rollout percentage. + + `unknown` covers drains enqueued before these arguments deployed. + """ + return { + "dispatcher": dispatcher or "unknown", + "mode": mode or "unknown", + } + + def _record_dispatch( *, dispatcher: Dispatcher, - mode: str, + mode: DispatchMode, drain: DispatchOutcome, mailbox_name: str, claimed: int = 0, @@ -313,20 +339,25 @@ def _claim_and_dispatch( the mailbox head is due again and another dispatcher can start a second, overlapping drain — duplicating deliveries and breaking mailbox ordering. - `dispatcher` only tags the dispatch metrics; both callers claim identically. + `dispatcher` tags this dispatch and is forwarded to the drain so its + deliveries carry the same attribution; both callers claim identically. """ claimed = _claim_mailbox_batch(head_id, mailbox_name, only_if_head_due=True) if not claimed: return DispatchOutcome.NOT_DUE if claimed >= PARALLEL_DRAIN_THRESHOLD: - drain_mailbox_parallel.delay(head_id, claimed_count=claimed) + drain_mailbox_parallel.delay( + head_id, claimed_count=claimed, dispatcher=dispatcher, mode=DispatchMode.CLAIM + ) outcome = DispatchOutcome.PARALLEL else: - drain_mailbox.delay(head_id, claimed_count=claimed) + drain_mailbox.delay( + head_id, claimed_count=claimed, dispatcher=dispatcher, mode=DispatchMode.CLAIM + ) outcome = DispatchOutcome.SEQUENTIAL _record_dispatch( dispatcher=dispatcher, - mode="claim", + mode=DispatchMode.CLAIM, drain=outcome, mailbox_name=mailbox_name, claimed=claimed, @@ -428,10 +459,15 @@ def _maybe_trigger_drain_lease(mailbox_name: str) -> None: ) return head_id = head[0] - drain_mailbox.delay(head_id, mailbox_name=mailbox_name) + drain_mailbox.delay( + head_id, + mailbox_name=mailbox_name, + dispatcher=Dispatcher.PUSH, + mode=DispatchMode.LEASE, + ) _record_dispatch( dispatcher=Dispatcher.PUSH, - mode="lease", + mode=DispatchMode.LEASE, drain=DispatchOutcome.SEQUENTIAL, mailbox_name=mailbox_name, ) @@ -542,14 +578,18 @@ def schedule_webhook_delivery() -> None: pass claimed = _claim_mailbox_batch(record["id"], mailbox_name) if claimed >= PARALLEL_DRAIN_THRESHOLD: - drain_mailbox_parallel.delay(record["id"]) + drain_mailbox_parallel.delay( + record["id"], dispatcher=Dispatcher.SCHEDULER, mode=DispatchMode.LEASE + ) drain = DispatchOutcome.PARALLEL else: - drain_mailbox.delay(record["id"]) + drain_mailbox.delay( + record["id"], dispatcher=Dispatcher.SCHEDULER, mode=DispatchMode.LEASE + ) drain = DispatchOutcome.SEQUENTIAL _record_dispatch( dispatcher=Dispatcher.SCHEDULER, - mode="lease", + mode=DispatchMode.LEASE, drain=drain, mailbox_name=mailbox_name, claimed=claimed, @@ -578,7 +618,11 @@ def schedule_webhook_delivery() -> None: silo_mode=SiloMode.CONTROL, ) def drain_mailbox( - payload_id: int, mailbox_name: str | None = None, claimed_count: int | None = None + payload_id: int, + mailbox_name: str | None = None, + claimed_count: int | None = None, + dispatcher: str | None = None, + mode: str | None = None, ) -> None: """ Deliver webhooks from the mailbox that `payload_id` is the head of. @@ -596,7 +640,11 @@ def drain_mailbox( `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. + + `dispatcher` and `mode` carry the enqueueing dispatcher's attribution onto + every delivery outcome this drain records (see `_dispatch_tags`). """ + dispatch_tags = _dispatch_tags(dispatcher, mode) try: payload = WebhookPayload.objects.get(id=payload_id) except WebhookPayload.DoesNotExist: @@ -604,7 +652,11 @@ def drain_mailbox( # and let the other process continue, or a future process. metrics.incr( "hybridcloud.deliver_webhooks.delivery", - tags={"outcome": "race", "provider": _provider_from_mailbox(mailbox_name)}, + tags={ + **dispatch_tags, + "outcome": "race", + "provider": _provider_from_mailbox(mailbox_name), + }, ) logger.info("deliver_webhook.potential_race", extra={"id": payload_id}) # Release the drain lock if we know the mailbox name. Otherwise the lock is @@ -639,7 +691,11 @@ def drain_mailbox( ) metrics.incr( "hybridcloud.deliver_webhooks.delivery", - tags={"outcome": "delivery_deadline", "provider": _provider_tag(payload)}, + tags={ + **dispatch_tags, + "outcome": "delivery_deadline", + "provider": _provider_tag(payload), + }, ) break @@ -661,13 +717,17 @@ def drain_mailbox( if mailbox_name and options.get("hybridcloud.webhookpayload.push_drain_trigger"): _refresh_drain_lock(payload.mailbox_name) try: - if deliver_message(record): + if deliver_message(record, dispatch_tags=dispatch_tags): delivered += 1 except DeliveryFailed: failed += 1 metrics.incr( "hybridcloud.deliver_webhooks.delivery", - tags={"outcome": "retry", "provider": _provider_tag(record)}, + tags={ + **dispatch_tags, + "outcome": "retry", + "provider": _provider_tag(record), + }, ) if not skip_on_failure: # For providers that require strict ordering, stop on the @@ -714,7 +774,7 @@ def drain_mailbox( ) metrics.incr( "hybridcloud.deliver_webhooks.delivery", - tags={"outcome": "claim_exhausted"}, + tags={**dispatch_tags, "outcome": "claim_exhausted"}, ) return finally: @@ -724,7 +784,7 @@ def drain_mailbox( _release_drain_lock(mailbox_name) -def _discard_if_stale(payload: WebhookPayload) -> bool: +def _discard_if_stale(payload: WebhookPayload, *, dispatch_tags: Mapping[str, str]) -> bool: """ Discard the payload when it is older than MAX_DELIVERY_AGE; returns whether it was discarded. Runs per record inside the drain walk, so stale rows @@ -739,7 +799,7 @@ def _discard_if_stale(payload: WebhookPayload) -> bool: # wants an exact total rather than an estimated rate. metrics.incr( "hybridcloud.deliver_webhooks.delivery", - tags={"outcome": "max_age", "provider": _provider_tag(payload)}, + tags={**dispatch_tags, "outcome": "max_age", "provider": _provider_tag(payload)}, sample_rate=1.0, ) logger.warning("deliver_webhook.max_age_discard", extra={**payload_data}) @@ -796,7 +856,7 @@ def _record_delivery_time_metrics(payload: WebhookPayload) -> None: def _handle_parallel_delivery_result( - payload_record: WebhookPayload, err: Exception | None + payload_record: WebhookPayload, err: Exception | None, *, dispatch_tags: Mapping[str, str] ) -> tuple[bool, bool]: """ Process one result from the parallel delivery threadpool. @@ -809,7 +869,11 @@ def _handle_parallel_delivery_result( payload_record.delete() metrics.incr( "hybridcloud.deliver_webhooks.delivery", - tags={"outcome": err.outcome, "provider": _provider_tag(payload_record)}, + tags={ + **dispatch_tags, + "outcome": err.outcome, + "provider": _provider_tag(payload_record), + }, ) return (False, False) if err: @@ -819,7 +883,11 @@ def _handle_parallel_delivery_result( # wants an exact total rather than an estimated rate. metrics.incr( "hybridcloud.deliver_webhooks.delivery", - tags={"outcome": "attempts_exceed", "provider": _provider_tag(payload_record)}, + tags={ + **dispatch_tags, + "outcome": "attempts_exceed", + "provider": _provider_tag(payload_record), + }, sample_rate=1.0, ) logger.warning( @@ -830,7 +898,11 @@ def _handle_parallel_delivery_result( else: metrics.incr( "hybridcloud.deliver_webhooks.delivery", - tags={"outcome": "retry", "provider": _provider_tag(payload_record)}, + tags={ + **dispatch_tags, + "outcome": "retry", + "provider": _provider_tag(payload_record), + }, ) payload_record.schedule_next_attempt() request_failed = True @@ -840,7 +912,7 @@ def _handle_parallel_delivery_result( _record_delivery_time_metrics(payload_record) metrics.incr( "hybridcloud.deliver_webhooks.delivery", - tags={"outcome": "ok", "provider": _provider_tag(payload_record)}, + tags={**dispatch_tags, "outcome": "ok", "provider": _provider_tag(payload_record)}, ) if timezone.now() - date_added >= SLOW_DELIVERY_THRESHOLD: logger.warning("deliver_webhook.slow_delivery", extra=payload_data) @@ -848,7 +920,7 @@ def _handle_parallel_delivery_result( def _run_parallel_delivery_batch( - mailbox_name: str, start_id: int, batch_size: int + mailbox_name: str, start_id: int, batch_size: int, *, dispatch_tags: Mapping[str, str] ) -> tuple[int, int, bool, int | None]: """ Run one batch of up to `batch_size` parallel deliveries for the mailbox. @@ -872,7 +944,9 @@ def _run_parallel_delivery_batch( # Stale rows are discarded in place of delivery, consuming claim budget # like any delivered row rather than being swept out from under the claim. - fresh_records = [record for record in records if not _discard_if_stale(record)] + fresh_records = [ + record for record in records if not _discard_if_stale(record, dispatch_tags=dispatch_tags) + ] delivered = 0 request_failed = False @@ -884,7 +958,7 @@ def _run_parallel_delivery_batch( for future in as_completed(futures): payload_record, err = future.result() batch_request_failed, should_reraise = _handle_parallel_delivery_result( - payload_record, err + payload_record, err, dispatch_tags=dispatch_tags ) request_failed = request_failed or batch_request_failed if should_reraise and err is not None: @@ -902,7 +976,11 @@ def _run_parallel_delivery_batch( silo_mode=SiloMode.CONTROL, ) def drain_mailbox_parallel( - payload_id: int, mailbox_name: str | None = None, claimed_count: int | None = None + payload_id: int, + mailbox_name: str | None = None, + claimed_count: int | None = None, + dispatcher: str | None = None, + mode: str | None = None, ) -> None: """ Deliver messages from a mailbox in small parallel batches. @@ -924,7 +1002,11 @@ def drain_mailbox_parallel( `mailbox_name` is accepted for symmetry with `drain_mailbox`; no current dispatcher passes it (lease triggers only dispatch sequential drains), but a caller that does owns the drain lock and gets it refreshed and released. + + `dispatcher` and `mode` carry the enqueueing dispatcher's attribution onto + every delivery outcome this drain records (see `_dispatch_tags`). """ + dispatch_tags = _dispatch_tags(dispatcher, mode) try: payload = WebhookPayload.objects.get(id=payload_id) except WebhookPayload.DoesNotExist: @@ -932,7 +1014,11 @@ def drain_mailbox_parallel( # and let the other process continue, or a future process. metrics.incr( "hybridcloud.deliver_webhooks.delivery", - tags={"outcome": "race", "provider": _provider_from_mailbox(mailbox_name)}, + tags={ + **dispatch_tags, + "outcome": "race", + "provider": _provider_from_mailbox(mailbox_name), + }, ) logger.info("deliver_webhook_parallel.potential_race", extra={"id": payload_id}) if mailbox_name and options.get("hybridcloud.webhookpayload.push_drain_trigger"): @@ -960,13 +1046,17 @@ def drain_mailbox_parallel( logger.info("deliver_webhook_parallel.delivery_deadline", extra=extra) metrics.incr( "hybridcloud.deliver_webhooks.delivery", - tags={"outcome": "delivery_deadline", "provider": _provider_tag(payload)}, + tags={ + **dispatch_tags, + "outcome": "delivery_deadline", + "provider": _provider_tag(payload), + }, ) break batch_size = worker_threads if remaining is None else min(worker_threads, remaining) attempted, delivered_batch, request_failed, next_id = _run_parallel_delivery_batch( - payload.mailbox_name, current_id, batch_size + payload.mailbox_name, current_id, batch_size, dispatch_tags=dispatch_tags ) delivered += delivered_batch extra["delivered"] = delivered @@ -988,7 +1078,7 @@ def drain_mailbox_parallel( logger.debug("deliver_webhook_parallel.claim_exhausted", extra=extra) metrics.incr( "hybridcloud.deliver_webhooks.delivery", - tags={"outcome": "claim_exhausted"}, + tags={**dispatch_tags, "outcome": "claim_exhausted"}, ) return @@ -1012,7 +1102,7 @@ def deliver_message_parallel(payload: WebhookPayload) -> tuple[WebhookPayload, E return (payload, err) -def deliver_message(payload: WebhookPayload) -> bool: +def deliver_message(payload: WebhookPayload, *, dispatch_tags: Mapping[str, str]) -> bool: """ Deliver a message if it still has delivery attempts remaining and is not stale. @@ -1027,13 +1117,17 @@ def deliver_message(payload: WebhookPayload) -> bool: # Unsampled: see the parallel discard path above. metrics.incr( "hybridcloud.deliver_webhooks.delivery", - tags={"outcome": "attempts_exceed", "provider": _provider_tag(payload)}, + tags={ + **dispatch_tags, + "outcome": "attempts_exceed", + "provider": _provider_tag(payload), + }, sample_rate=1.0, ) logger.warning("deliver_webhook.discard", extra={**payload_data}) return False - if _discard_if_stale(payload): + if _discard_if_stale(payload, dispatch_tags=dispatch_tags): return False payload.schedule_next_attempt() @@ -1045,7 +1139,7 @@ def deliver_message(payload: WebhookPayload) -> bool: payload.delete() metrics.incr( "hybridcloud.deliver_webhooks.delivery", - tags={"outcome": err.outcome, "provider": _provider_tag(payload)}, + tags={**dispatch_tags, "outcome": err.outcome, "provider": _provider_tag(payload)}, ) return False date_added = payload.date_added @@ -1055,7 +1149,7 @@ def deliver_message(payload: WebhookPayload) -> bool: logger.warning("deliver_webhook.slow_delivery", extra=payload_data) metrics.incr( "hybridcloud.deliver_webhooks.delivery", - tags={"outcome": "ok", "provider": _provider_tag(payload)}, + tags={**dispatch_tags, "outcome": "ok", "provider": _provider_tag(payload)}, ) return True diff --git a/tests/sentry/hybridcloud/tasks/test_deliver_webhooks.py b/tests/sentry/hybridcloud/tasks/test_deliver_webhooks.py index 796ecbaf0a76..18f261f115cd 100644 --- a/tests/sentry/hybridcloud/tasks/test_deliver_webhooks.py +++ b/tests/sentry/hybridcloud/tasks/test_deliver_webhooks.py @@ -20,6 +20,7 @@ PARALLEL_DRAIN_THRESHOLD, SLOW_DELIVERY_THRESHOLD, Dispatcher, + DispatchMode, _claim_and_dispatch, _use_claim_dispatch, drain_mailbox, @@ -37,6 +38,10 @@ cell_config = [Cell("us", 1, "http://us.testserver")] +# Drains invoked directly carry no dispatcher attribution; `_dispatch_tags` +# tags them rather than omitting the keys, so the tag stays queryable. +UNATTRIBUTED = {"dispatcher": "unknown", "mode": "unknown"} + CLAIM_MODE_OPTIONS = { "hybridcloud.webhookpayload.push_drain_trigger": True, "hybridcloud.webhookpayload.claim_dispatch_rollout": 1.0, @@ -86,7 +91,9 @@ def test_schedule_one_mailbox_multiple_messages(self, mock_deliver: MagicMock) - ) schedule_webhook_delivery() assert mock_deliver.delay.call_count == 1 - mock_deliver.delay.assert_called_with(webhook_one.id) + mock_deliver.delay.assert_called_with( + webhook_one.id, dispatcher=Dispatcher.SCHEDULER, mode=DispatchMode.LEASE + ) @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") def test_schedule_mailbox_scheduled_later(self, mock_deliver: MagicMock) -> None: @@ -101,7 +108,9 @@ def test_schedule_mailbox_scheduled_later(self, mock_deliver: MagicMock) -> None ) schedule_webhook_delivery() assert mock_deliver.delay.call_count == 1 - mock_deliver.delay.assert_called_with(webhook_one.id) + mock_deliver.delay.assert_called_with( + webhook_one.id, dispatcher=Dispatcher.SCHEDULER, mode=DispatchMode.LEASE + ) @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") def test_schedule_updates_mailbox_attributes(self, mock_deliver: MagicMock) -> None: @@ -124,7 +133,9 @@ def test_schedule_updates_mailbox_attributes(self, mock_deliver: MagicMock) -> N assert webhook_two.schedule_for > timezone.now() assert mock_deliver.delay.call_count == 1 - mock_deliver.delay.assert_called_with(webhook_one.id) + mock_deliver.delay.assert_called_with( + webhook_one.id, dispatcher=Dispatcher.SCHEDULER, mode=DispatchMode.LEASE + ) @responses.activate @override_cells(cell_config) @@ -218,7 +229,9 @@ def test_claim_and_dispatch_claims_in_a_single_query( ) assert outcome == "sequential" - mock_drain.delay.assert_called_once_with(webhook.id, claimed_count=1) + mock_drain.delay.assert_called_once_with( + webhook.id, claimed_count=1, dispatcher=Dispatcher.SCHEDULER, mode=DispatchMode.CLAIM + ) queries = [ q["sql"] for q in ctx.captured_queries @@ -1389,7 +1402,7 @@ def test_delivery_tagged_with_provider(self, mock_metrics: MagicMock) -> None: drain_mailbox(webhook.id) assert self.tags_for(mock_metrics, "hybridcloud.deliver_webhooks.delivery") == [ - {"outcome": "ok", "provider": "github"} + {**UNATTRIBUTED, "outcome": "ok", "provider": "github"} ] @responses.activate @@ -1431,7 +1444,7 @@ def test_dropped_outcomes_tagged_with_provider(self, mock_metrics: MagicMock) -> drain_mailbox(webhook.id) assert self.tags_for(mock_metrics, "hybridcloud.deliver_webhooks.delivery") == [ - {"outcome": "conflict", "provider": "github"} + {**UNATTRIBUTED, "outcome": "conflict", "provider": "github"} ] @responses.activate @@ -1451,7 +1464,7 @@ def test_parallel_dropped_outcome_tagged_with_provider(self, mock_metrics: Magic drain_mailbox_parallel(webhook.id) assert self.tags_for(mock_metrics, "hybridcloud.deliver_webhooks.delivery") == [ - {"outcome": "dropped_4xx", "provider": "github"} + {**UNATTRIBUTED, "outcome": "dropped_4xx", "provider": "github"} ] @responses.activate @@ -1471,7 +1484,7 @@ def test_provider_falls_back_to_unknown(self, mock_metrics: MagicMock) -> None: drain_mailbox(webhook.id) assert self.tags_for(mock_metrics, "hybridcloud.deliver_webhooks.delivery") == [ - {"outcome": "ok", "provider": "unknown"} + {**UNATTRIBUTED, "outcome": "ok", "provider": "unknown"} ] @override_options({"hybridcloud.webhookpayload.push_drain_trigger": True}) @@ -1533,8 +1546,8 @@ def test_retry_tagged_from_failing_record_not_mailbox_head( drain_mailbox(head.id) assert self.tags_for(mock_metrics, "hybridcloud.deliver_webhooks.delivery") == [ - {"outcome": "ok", "provider": "github"}, - {"outcome": "retry", "provider": "unknown"}, + {**UNATTRIBUTED, "outcome": "ok", "provider": "github"}, + {**UNATTRIBUTED, "outcome": "retry", "provider": "unknown"}, ] @override_options({"hybridcloud.webhookpayload.push_drain_trigger": True}) @@ -1543,7 +1556,7 @@ def test_race_provider_from_mailbox_name(self, mock_metrics: MagicMock) -> None: drain_mailbox(999999, mailbox_name="gitlab:7") assert self.tags_for(mock_metrics, "hybridcloud.deliver_webhooks.delivery") == [ - {"outcome": "race", "provider": "gitlab"} + {**UNATTRIBUTED, "outcome": "race", "provider": "gitlab"} ] @@ -1578,6 +1591,8 @@ def test_push_claim_dispatch_attributed_to_push( maybe_trigger_drain(webhook.mailbox_name) + assert mock_drain.delay.call_args.kwargs["dispatcher"] == Dispatcher.PUSH + assert mock_drain.delay.call_args.kwargs["mode"] == DispatchMode.CLAIM assert self.dispatch_tags(mock_metrics) == [ { "dispatcher": "push", @@ -1611,6 +1626,8 @@ def test_push_lease_dispatch_claims_zero( maybe_trigger_drain(webhook.mailbox_name) + assert mock_drain.delay.call_args.kwargs["dispatcher"] == Dispatcher.PUSH + assert mock_drain.delay.call_args.kwargs["mode"] == DispatchMode.LEASE assert self.dispatch_tags(mock_metrics) == [ { "dispatcher": "push", @@ -1640,6 +1657,8 @@ def test_scheduler_lease_dispatch_attributed_to_scheduler( schedule_webhook_delivery() + assert mock_drain.delay.call_args.kwargs["dispatcher"] == Dispatcher.SCHEDULER + assert mock_drain.delay.call_args.kwargs["mode"] == DispatchMode.LEASE assert self.dispatch_tags(mock_metrics) == [ { "dispatcher": "scheduler", @@ -1673,6 +1692,8 @@ def test_scheduler_claim_dispatch_reports_batch_depth( schedule_webhook_delivery() + assert mock_drain_parallel.delay.call_args.kwargs["dispatcher"] == Dispatcher.SCHEDULER + assert mock_drain_parallel.delay.call_args.kwargs["mode"] == DispatchMode.CLAIM assert self.dispatch_tags(mock_metrics) == [ { "dispatcher": "scheduler", @@ -1694,6 +1715,97 @@ def test_scheduler_claim_dispatch_reports_batch_depth( ] +@control_silo_test +class DeliveryDispatchTagTest(TestCase): + """ + Delivery outcomes carry the attribution of the drain that produced them. + Without it a partial claim rollout is only readable as a shift in a global + total, which that total's own variance swamps at low percentages. + """ + + def delivery_tags(self, mock_metrics: MagicMock) -> list[dict[str, str]]: + return [ + c[1].get("tags", {}) + for c in mock_metrics.incr.call_args_list + if c[0][0] == "hybridcloud.deliver_webhooks.delivery" + ] + + @responses.activate + @override_cells(cell_config) + @patch("sentry.hybridcloud.tasks.deliver_webhooks.metrics") + def test_delivery_carries_dispatch_attribution(self, mock_metrics: MagicMock) -> None: + responses.add( + responses.POST, + "http://us.testserver/extensions/github/webhook/", + status=200, + body="", + ) + webhook = self.create_webhook_payload( + mailbox_name="github:123", cell_name="us", provider="github" + ) + + drain_mailbox( + webhook.id, + claimed_count=1, + dispatcher=Dispatcher.SCHEDULER, + mode=DispatchMode.CLAIM, + ) + + # `claim_exhausted` has no provider tag, so it is the outcome most likely + # to be missed when attribution is added; assert it alongside the delivery. + assert self.delivery_tags(mock_metrics) == [ + { + "dispatcher": "scheduler", + "mode": "claim", + "outcome": "ok", + "provider": "github", + }, + {"dispatcher": "scheduler", "mode": "claim", "outcome": "claim_exhausted"}, + ] + + @responses.activate + @patch("sentry.hybridcloud.tasks.deliver_webhooks.metrics") + def test_race_outcome_carries_mode(self, mock_metrics: MagicMock) -> None: + # Races are the rollout's primary benefit signal, so this is the outcome + # that most needs to be comparable between the claim and lease cohorts. + drain_mailbox(99, dispatcher=Dispatcher.PUSH, mode=DispatchMode.CLAIM) + + assert self.delivery_tags(mock_metrics) == [ + { + "dispatcher": "push", + "mode": "claim", + "outcome": "race", + "provider": "unknown", + } + ] + + @responses.activate + @patch("sentry.hybridcloud.tasks.deliver_webhooks.metrics") + def test_parallel_drain_carries_dispatch_attribution(self, mock_metrics: MagicMock) -> None: + drain_mailbox_parallel(99, dispatcher=Dispatcher.SCHEDULER, mode=DispatchMode.LEASE) + + assert self.delivery_tags(mock_metrics) == [ + { + "dispatcher": "scheduler", + "mode": "lease", + "outcome": "race", + "provider": "unknown", + } + ] + + @responses.activate + @patch("sentry.hybridcloud.tasks.deliver_webhooks.metrics") + def test_drain_enqueued_before_deploy_tags_unknown(self, mock_metrics: MagicMock) -> None: + # Tasks already queued when this deploys arrive without the arguments. The + # keys must still be emitted: a tag absent from some series breaks grouping + # rather than showing a gap. + drain_mailbox(99) + + assert self.delivery_tags(mock_metrics) == [ + {**UNATTRIBUTED, "outcome": "race", "provider": "unknown"} + ] + + @control_silo_test class PushTriggerTest(TestCase): @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") @@ -1701,7 +1813,9 @@ class PushTriggerTest(TestCase): def test_push_trigger_enqueues_drain_for_idle_mailbox(self, mock_drain: MagicMock) -> None: webhook = self.create_webhook_payload(mailbox_name="github:123", cell_name="us") maybe_trigger_drain(webhook.mailbox_name) - mock_drain.delay.assert_called_once_with(webhook.id, claimed_count=1) + mock_drain.delay.assert_called_once_with( + webhook.id, claimed_count=1, dispatcher=Dispatcher.PUSH, mode=DispatchMode.CLAIM + ) # The batch is claimed before dispatch; the claim is what keeps other # dispatchers off the mailbox while the drain runs. webhook.refresh_from_db() @@ -1741,7 +1855,9 @@ def test_push_trigger_drains_from_mailbox_head_not_new_payload( # Trigger with the newer webhook's ID, as get_response_from_webhookpayload does maybe_trigger_drain(newer_webhook.mailbox_name) # Must drain from the head of the mailbox so the older payload is not skipped - mock_drain.delay.assert_called_once_with(older_webhook.id, claimed_count=2) + mock_drain.delay.assert_called_once_with( + older_webhook.id, claimed_count=2, dispatcher=Dispatcher.PUSH, mode=DispatchMode.CLAIM + ) @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") @override_options(CLAIM_MODE_OPTIONS) @@ -1823,7 +1939,9 @@ def test_scheduler_skips_locked_mailboxes(self, mock_drain: MagicMock) -> None: # Only mailbox B should have been scheduled assert mock_drain.delay.call_count == 1 - mock_drain.delay.assert_called_once_with(webhook_b.id) + mock_drain.delay.assert_called_once_with( + webhook_b.id, dispatcher=Dispatcher.SCHEDULER, mode=DispatchMode.LEASE + ) @override_options(CLAIM_MODE_OPTIONS) @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") @@ -1874,7 +1992,9 @@ def test_push_trigger_fires_immediately_after_drain_completes( # trigger a fresh drain right away. webhook_two = self.create_webhook_payload(mailbox_name="github:123", cell_name="us") maybe_trigger_drain(webhook_two.mailbox_name) - mock_drain.delay.assert_called_once_with(webhook_two.id, claimed_count=1) + mock_drain.delay.assert_called_once_with( + webhook_two.id, claimed_count=1, dispatcher=Dispatcher.PUSH, mode=DispatchMode.CLAIM + ) @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") @override_options({"hybridcloud.webhookpayload.push_drain_trigger": True}) @@ -1935,7 +2055,10 @@ def test_push_trigger_uses_parallel_drain_for_deep_mailbox( # A mailbox this deep is behind; the sequential drain would work it off at # one in-flight request for its whole run. mock_drain_parallel.delay.assert_called_once_with( - records[0].id, claimed_count=PARALLEL_DRAIN_THRESHOLD + records[0].id, + claimed_count=PARALLEL_DRAIN_THRESHOLD, + dispatcher=Dispatcher.PUSH, + mode=DispatchMode.CLAIM, ) mock_drain.delay.assert_not_called() @@ -1951,7 +2074,10 @@ def test_push_trigger_uses_sequential_drain_for_shallow_mailbox( # One record short of the threshold keeps strict ordering. mock_drain.delay.assert_called_once_with( - records[0].id, claimed_count=PARALLEL_DRAIN_THRESHOLD - 1 + records[0].id, + claimed_count=PARALLEL_DRAIN_THRESHOLD - 1, + dispatcher=Dispatcher.PUSH, + mode=DispatchMode.CLAIM, ) mock_drain_parallel.delay.assert_not_called() @@ -2001,7 +2127,12 @@ def test_lease_mode_trigger_hands_lock_to_drain(self, mock_drain: MagicMock) -> # The drain receives mailbox_name and owns the lock for its whole run; # the trigger must not release it. - mock_drain.delay.assert_called_once_with(webhook.id, mailbox_name=webhook.mailbox_name) + mock_drain.delay.assert_called_once_with( + webhook.id, + mailbox_name=webhook.mailbox_name, + dispatcher=Dispatcher.PUSH, + mode=DispatchMode.LEASE, + ) assert cache.get(f"wh:drain_active:{webhook.mailbox_name}") == 1 # No claim in lease mode: the batch stays due for the scheduler's view. webhook.refresh_from_db() From ac91fcb698a9919a416d885a765b0e200ea7ea8d Mon Sep 17 00:00:00 2001 From: Ivan Dlugos Date: Fri, 14 Aug 2026 08:09:49 +0200 Subject: [PATCH 2/2] feat(hybridcloud): Attribute delivery latency to its dispatcher and mode The outcome counter distinguishes the claim and lease cohorts, but latency is the quantity the claim regime is meant to move, and it was only readable in aggregate. A global p50 cannot show a cohort effect while the rollout is a small share of traffic. `delivery_time_ms` is a native Datadog distribution rather than the usual histogram, so percentiles are queryable per tag combination. --- .../hybridcloud/tasks/deliver_webhooks.py | 17 ++++- .../tasks/test_deliver_webhooks.py | 74 +++++++++++++++++++ 2 files changed, 87 insertions(+), 4 deletions(-) diff --git a/src/sentry/hybridcloud/tasks/deliver_webhooks.py b/src/sentry/hybridcloud/tasks/deliver_webhooks.py index fc482e97ef8a..b1dab6436159 100644 --- a/src/sentry/hybridcloud/tasks/deliver_webhooks.py +++ b/src/sentry/hybridcloud/tasks/deliver_webhooks.py @@ -839,10 +839,19 @@ def _get_github_delivery_time_tags(payload: WebhookPayload) -> dict[str, str]: return {"github_event_and_action": f"{event_type}.{action}"} -def _record_delivery_time_metrics(payload: WebhookPayload) -> None: - """Record delivery time metrics for a successfully delivered webhook payload.""" +def _record_delivery_time_metrics( + payload: WebhookPayload, *, dispatch_tags: Mapping[str, str] +) -> None: + """Record delivery time metrics for a successfully delivered webhook payload. + + Measured from `date_added`, so this is queue wait plus retry backoff plus the + request itself — the span dispatch is meant to shorten. It carries the same + attribution as the outcome counter so latency can be compared per dispatcher + and per regime rather than only in aggregate. + """ duration = timezone.now() - payload.date_added tags = { + **dispatch_tags, "region_sent_to": payload.cell_name, "provider": _provider_tag(payload), } | _get_github_delivery_time_tags(payload) @@ -909,7 +918,7 @@ def _handle_parallel_delivery_result( return (request_failed, not isinstance(err, DeliveryFailed)) date_added = payload_record.date_added payload_record.delete() - _record_delivery_time_metrics(payload_record) + _record_delivery_time_metrics(payload_record, dispatch_tags=dispatch_tags) metrics.incr( "hybridcloud.deliver_webhooks.delivery", tags={**dispatch_tags, "outcome": "ok", "provider": _provider_tag(payload_record)}, @@ -1144,7 +1153,7 @@ def deliver_message(payload: WebhookPayload, *, dispatch_tags: Mapping[str, str] return False date_added = payload.date_added payload.delete() - _record_delivery_time_metrics(payload) + _record_delivery_time_metrics(payload, dispatch_tags=dispatch_tags) if timezone.now() - date_added >= SLOW_DELIVERY_THRESHOLD: logger.warning("deliver_webhook.slow_delivery", extra=payload_data) metrics.incr( diff --git a/tests/sentry/hybridcloud/tasks/test_deliver_webhooks.py b/tests/sentry/hybridcloud/tasks/test_deliver_webhooks.py index 18f261f115cd..21b68394ec90 100644 --- a/tests/sentry/hybridcloud/tasks/test_deliver_webhooks.py +++ b/tests/sentry/hybridcloud/tasks/test_deliver_webhooks.py @@ -1145,6 +1145,10 @@ def test_delivery_time_metrics_cell_sent_to(self, mock_metrics: MagicMock) -> No assert tags.get("region_sent_to") == "us" # Rows predating the provider column still drain through here. assert tags.get("provider") == "unknown" + # A drain with no dispatch arguments still emits the attribution keys; a + # tag missing from some series breaks grouping rather than showing a gap. + assert tags.get("dispatcher") == "unknown" + assert tags.get("mode") == "unknown" @responses.activate @override_cells(cell_config) @@ -1730,6 +1734,13 @@ def delivery_tags(self, mock_metrics: MagicMock) -> list[dict[str, str]]: if c[0][0] == "hybridcloud.deliver_webhooks.delivery" ] + def delivery_time_tags(self, mock_metrics: MagicMock) -> list[dict[str, str]]: + return [ + c[1].get("tags", {}) + for c in mock_metrics.distribution.call_args_list + if c[0][0] == "hybridcloud.deliver_webhooks.delivery_time_ms" + ] + @responses.activate @override_cells(cell_config) @patch("sentry.hybridcloud.tasks.deliver_webhooks.metrics") @@ -1763,6 +1774,69 @@ def test_delivery_carries_dispatch_attribution(self, mock_metrics: MagicMock) -> {"dispatcher": "scheduler", "mode": "claim", "outcome": "claim_exhausted"}, ] + @responses.activate + @override_cells(cell_config) + @patch("sentry.hybridcloud.tasks.deliver_webhooks.metrics") + def test_delivery_time_carries_dispatch_attribution(self, mock_metrics: MagicMock) -> None: + # Latency is the quantity the claim regime is meant to move, so it needs + # the same attribution as the counter to be comparable between cohorts. + responses.add( + responses.POST, + "http://us.testserver/extensions/github/webhook/", + status=200, + body="", + ) + webhook = self.create_webhook_payload( + mailbox_name="github:123", cell_name="us", provider="github" + ) + + drain_mailbox( + webhook.id, + claimed_count=1, + dispatcher=Dispatcher.SCHEDULER, + mode=DispatchMode.CLAIM, + ) + + assert self.delivery_time_tags(mock_metrics) == [ + { + "dispatcher": "scheduler", + "mode": "claim", + "region_sent_to": "us", + "provider": "github", + } + ] + + @responses.activate + @override_cells(cell_config) + @patch("sentry.hybridcloud.tasks.deliver_webhooks.metrics") + def test_parallel_delivery_time_carries_dispatch_attribution( + self, mock_metrics: MagicMock + ) -> None: + # The parallel path records latency from its own callsite, so it can lose + # attribution independently of the sequential one. + responses.add( + responses.POST, + "http://us.testserver/extensions/github/webhook/", + status=200, + body="", + ) + records = create_payloads(2, "github:123", provider="github") + + drain_mailbox_parallel( + records[0].id, + claimed_count=2, + dispatcher=Dispatcher.PUSH, + mode=DispatchMode.CLAIM, + ) + + expected = { + "dispatcher": "push", + "mode": "claim", + "region_sent_to": "us", + "provider": "github", + } + assert self.delivery_time_tags(mock_metrics) == [expected, expected] + @responses.activate @patch("sentry.hybridcloud.tasks.deliver_webhooks.metrics") def test_race_outcome_carries_mode(self, mock_metrics: MagicMock) -> None: