diff --git a/src/sentry/hybridcloud/tasks/deliver_webhooks.py b/src/sentry/hybridcloud/tasks/deliver_webhooks.py index f9e1287922ff..b1dab6436159 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}) @@ -779,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) @@ -796,7 +865,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 +878,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 +892,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,17 +907,21 @@ 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 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={"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 +929,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 +953,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 +967,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 +985,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 +1011,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 +1023,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 +1055,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 +1087,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 +1111,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 +1126,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,17 +1148,17 @@ 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 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( "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..21b68394ec90 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 @@ -1132,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) @@ -1389,7 +1406,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 +1448,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 +1468,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 +1488,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 +1550,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 +1560,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 +1595,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 +1630,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 +1661,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 +1696,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 +1719,167 @@ 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" + ] + + 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") + 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 + @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: + # 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 +1887,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 +1929,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 +2013,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 +2066,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 +2129,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 +2148,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 +2201,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()