diff --git a/src/sentry/hybridcloud/webhook_mailbox_sizing.py b/src/sentry/hybridcloud/webhook_mailbox_sizing.py new file mode 100644 index 000000000000..3278aea57edd --- /dev/null +++ b/src/sentry/hybridcloud/webhook_mailbox_sizing.py @@ -0,0 +1,157 @@ +""" +Sizing an integration's webhook mailbox split from the rate it is currently sending. + +Only where the delivery side may reorder: the divisor is the key-to-mailbox map, and +a provider delivered in order cannot have that map move under it. +""" + +from __future__ import annotations + +import logging +from dataclasses import replace +from time import time + +from django.conf import settings +from redis.exceptions import RedisError + +from sentry import options +from sentry.hybridcloud.mailbox import MailboxName +from sentry.utils import redis + +logger = logging.getLogger(__name__) + +WINDOW_SECONDS = 15 * 60 +"""Long enough that a lull between bursts does not read as a quiet integration, +short enough that a burst raises the count within minutes.""" + +SHARD_SECONDS = 3 * 60 +"""Width of one counter: how coarsely the window rolls, how far the estimate +undercounts while the current shard fills, and how many keys one read touches. + +Nothing derives three minutes -- it is a fifth of the window. Minute shards would +read fifteen keys to be accurate within a fifteenth; five-minute shards three keys +to be accurate within a third. +""" + +SHARD_COUNT = WINDOW_SECONDS // SHARD_SECONDS + +SHARD_TTL_SECONDS = WINDOW_SECONDS + SHARD_SECONDS +"""A real bound despite being reset on every write: a shard takes writes only while +it is the current one.""" + +STRICT_BUCKET_COUNT = 10 +"""Width for a provider delivered in order. Every change of divisor re-maps keys, +which would leave one issue's backlog draining concurrently with its next +payloads.""" + + +def mailbox_bucket_count(mailbox: MailboxName) -> int: + """How many sub-mailboxes to spread `mailbox`'s bucket keys over. + + Counts this payload against the window, so call it once per payload queued. A + strictly ordered provider is not counted: nothing would read the result. + """ + if not _tolerates_reordering(mailbox.provider): + return STRICT_BUCKET_COUNT + return _count_for_payloads(_record_and_read_window(_rate_counter_key(mailbox))) + + +def _payloads_per_mailbox() -> int: + """Depth a mailbox reaches before its split widens. + + A drain delivers `worker_threads` payloads at once, so the depth is a whole + number of those, and raising delivery concurrency narrows the split rather than + leaving it where it was. + + Floored at one because both options are automator-modifiable and neither is + validated: the drain reads a zero `worker_threads` as one rather than rejecting + it, so a zero must not reach the division here either. + """ + return max( + 1, + options.get("hybridcloud.webhookpayload.payloads_per_thread") + * options.get("hybridcloud.webhookpayload.worker_threads"), + ) + + +def _max_buckets() -> int: + """Widest split allowed, floored to a power of two. + + The ladder `_count_for_payloads` climbs is doublings, and a cap off that ladder + would make a resize into it re-map nearly every key instead of half -- so the set + value is floored rather than trusted. + """ + configured = options.get("hybridcloud.webhookpayload.max_mailbox_buckets") + return 1 << (configured.bit_length() - 1) if configured >= 1 else 1 + + +def _tolerates_reordering(provider: str) -> bool: + """Read from the option the drain itself reads, so a provider earns a + rate-derived width only once it tolerates the re-mapping that width costs.""" + return provider in (options.get("hybridcloud.webhookpayload.skip_on_failure_providers") or ()) + + +def _rate_counter_key(mailbox: MailboxName) -> str: + """The mailbox name without the parts one split varies: the bucket, which the + split chooses, and the cell, since one mailbox is built for the whole fanout. + + Event type stays -- it separates mailboxes ahead of bucketing, so dropping it + would size one split for every event type at once. + """ + return str(replace(mailbox, cell=None, bucket=None)) + + +def _count_for_payloads(payloads: int | None) -> int: + """The widest split whose mailboxes would each still fill to the target depth, + given the payloads counted over the window. + + Powers of two because `key % 2n` puts a key where `key % n` did or n along, so a + doubling moves half the keys. Rounding down rather than to nearest is the + hysteresis: the count has to double to add a bucket. + + A window we could not read sizes to the cap -- an outage must not re-serialize + the integrations this exists to unserialize. + """ + if payloads is None: + return _max_buckets() + + mailboxes = payloads // _payloads_per_mailbox() + if mailboxes < 2: + return 1 + return min(1 << (mailboxes.bit_length() - 1), _max_buckets()) + + +def _record_and_read_window(counter_key: str) -> int | None: + """Count this payload and return the payloads over the window, or None when Redis + could not answer. + + A reply that does not destructure or coerce counts as not answering: this runs + before the payload row is written, so an exception escaping here would turn a + webhook we could still have queued into a 500. + + The current shard is still filling, so the sum runs low by up to one shard and + catches up -- the right direction, delaying a widening rather than forcing one. + """ + shard = int(time() // SHARD_SECONDS) + current_key = _shard_key(counter_key, shard) + older_keys = [_shard_key(counter_key, shard - i) for i in range(1, SHARD_COUNT)] + + try: + pipe = redis.redis_clusters.get(settings.SENTRY_RATE_LIMIT_REDIS_CLUSTER).pipeline() + pipe.incr(current_key) + pipe.expire(current_key, SHARD_TTL_SECONDS) + pipe.mget(older_keys) + current, _, older = pipe.execute() + return int(current) + sum(int(count) for count in older if count is not None) + except (RedisError, TypeError, ValueError, IndexError): + logger.exception( + "hybridcloud.webhook_mailbox_sizing.unavailable", + extra={"counter_key": counter_key}, + ) + return None + + +def _shard_key(counter_key: str, shard: int) -> str: + """Hash-tagged so one counter's shards share a slot and the window reads as a + single-node pipeline.""" + return f"whrate:{{{counter_key}}}:{shard}" diff --git a/src/sentry/integrations/middleware/hybrid_cloud/parser.py b/src/sentry/integrations/middleware/hybrid_cloud/parser.py index 86265a0383d7..85fc6b4b20c6 100644 --- a/src/sentry/integrations/middleware/hybrid_cloud/parser.py +++ b/src/sentry/integrations/middleware/hybrid_cloud/parser.py @@ -6,13 +6,11 @@ from typing import TYPE_CHECKING, Any, ClassVar import orjson -from django.core.cache import cache from django.http import HttpRequest, HttpResponse from django.http.response import HttpResponseBase from django.urls import ResolverMatch, resolve from rest_framework import status -from sentry.api.base import ONE_DAY from sentry.constants import ObjectStatus from sentry.hybridcloud.mailbox import MailboxName from sentry.hybridcloud.models.webhookpayload import DestinationType, WebhookPayload @@ -21,6 +19,7 @@ from sentry.hybridcloud.services.organization_mapping.model import RpcOrganizationMapping from sentry.hybridcloud.tasks.deliver_webhooks import maybe_trigger_drain from sentry.hybridcloud.webhook_event_types import MAILBOX_EVENT_TYPES +from sentry.hybridcloud.webhook_mailbox_sizing import mailbox_bucket_count from sentry.integrations.middleware.metrics import ( MiddlewareHaltReason, MiddlewareOperationEvent, @@ -31,7 +30,6 @@ from sentry.integrations.services.integration.model import RpcIntegration from sentry.killswitches import KillswitchConfig, get_killswitch_value, value_matches from sentry.logging.handlers import SamplingFilter -from sentry.ratelimits import backend as ratelimiter from sentry.silo.base import SiloLimit, SiloMode from sentry.silo.client import CellSiloClient, SiloClientError from sentry.types.cell import Cell, find_cells_for_org_mappings, get_cell_by_name @@ -79,17 +77,6 @@ class BaseRequestParser(ABC): webhook_identifier: ClassVar[WebhookProviderIdentifier] """The webhook provider identifier""" - mailbox_bucket_count: ClassVar[int] = 100 - """How many sub-mailboxes `mailbox_bucket_id` is spread over. - - Every mailbox costs a scheduler row and a dispatch slot, so splitting past what - the volume needs buys queue rows rather than parallelism. - """ - - always_bucket: ClassVar[bool] = False - """Split every integration's mailbox by `mailbox_bucket_id` instead of waiting - for it to exceed the hourly rate limit first.""" - def __init__(self, request: HttpRequest, response_handler: ResponseHandler): self.request = request self.match: ResolverMatch = resolve(self.request.path) @@ -326,64 +313,59 @@ def get_mailbox( that can be delivered in parallel. Requires the integration to implement `mailbox_bucket_id` - The cell is left for the fanout to add -- one mailbox is built for all of - them. + The event type is resolved before the bucket because the split is sized per + event type, and only the validated value may reach that: it is read out of a + body control has not verified. """ - return self._bucketed( - MailboxName( - provider=self.provider, - subject=str(integration.id), - event_type=self._mailbox_event_type(data), - ), - integration, - data, + mailbox = MailboxName( + provider=self.provider, + subject=str(integration.id), + event_type=self._mailbox_event_type(data), ) - - def _bucketed( - self, - mailbox: MailboxName, - integration: RpcIntegration | Integration, - data: dict[str, Any], - ) -> MailboxName: - """`mailbox` in a bucket, or unchanged where the integration is below the - volume that warrants buckets, or the payload carries no key to bucket it on.""" - if not self.always_bucket and not self._exceeds_bucketing_volume(integration): - self._record_mailbox_routing(bucketed=False, reason="under_volume_gate") + # Callers build the mailbox as an argument, so this runs before the shed check + # in get_response_from_webhookpayload. A shed payload is never queued, so it + # must not raise the rate that sizes the split -- nor pay the Redis write that + # shedding exists to avoid. + if self._should_shed(integration.id): return mailbox + return self._bucketed(mailbox, data) + def _bucketed(self, mailbox: MailboxName, data: dict[str, Any]) -> MailboxName: + """`mailbox` in a bucket, or unchanged where the payload does not get one. + + A keyless payload lands on the unsplit mailbox however wide the split is, so + it is left out of the rate that sizes it. + """ mailbox_bucket_id = self.mailbox_bucket_id(data) if mailbox_bucket_id is None: self._record_mailbox_routing(bucketed=False, reason="no_bucket_key") return mailbox - self._record_mailbox_routing(bucketed=True, reason="bucketed") + bucket_count = mailbox_bucket_count(mailbox) + if bucket_count == 1: + self._record_mailbox_routing(bucketed=False, reason="under_rate", buckets=1) + return mailbox - return mailbox.in_bucket(mailbox_bucket_id % self.mailbox_bucket_count) + self._record_mailbox_routing(bucketed=True, reason="bucketed", buckets=bucket_count) - def _exceeds_bucketing_volume(self, integration: RpcIntegration | Integration) -> bool: - # If we get fewer than 3000 in 1 hour we don't need to split into buckets - ratelimit_key = f"webhookpayload:{self.provider}:{integration.id}" - use_buckets_key = f"{ratelimit_key}:use_buckets" + return mailbox.in_bucket(mailbox_bucket_id % bucket_count) - if cache.get(use_buckets_key): - return True - if ratelimiter.is_limited(key=ratelimit_key, window=60 * 60, limit=3000): - # Once we have gone over the rate limit in a day, we use smaller - # buckets for the next day. - cache.set(use_buckets_key, 1, timeout=ONE_DAY) - return True - return False + def _record_mailbox_routing( + self, bucketed: bool, reason: str, buckets: int | None = None + ) -> None: + """`reason` is the full breakdown; `bucketed` stays for the dashboards on it. - def _record_mailbox_routing(self, bucketed: bool, reason: str) -> None: - """`reason` is the full breakdown; `bucketed` stays for the dashboards on it.""" - metrics.incr( - "hybridcloud.webhookpayload.mailbox_routing", - tags={ - "provider": self.provider, - "bucketed": "true" if bucketed else "false", - "reason": reason, - }, - ) + `buckets` is left off the path that never consults a count, so a query for + split width does not average in routing that has no width. + """ + tags = { + "provider": self.provider, + "bucketed": "true" if bucketed else "false", + "reason": reason, + } + if buckets is not None: + tags["buckets"] = str(buckets) + metrics.incr("hybridcloud.webhookpayload.mailbox_routing", tags=tags) def mailbox_bucket_id(self, data: dict[str, Any]) -> int | None: raise NotImplementedError( diff --git a/src/sentry/middleware/integrations/parsers/github.py b/src/sentry/middleware/integrations/parsers/github.py index 2c3ca9f053f7..1acedb57c6c4 100644 --- a/src/sentry/middleware/integrations/parsers/github.py +++ b/src/sentry/middleware/integrations/parsers/github.py @@ -66,7 +66,6 @@ class GithubRequestParser(BaseRequestParser): webhook_identifier = WebhookProviderIdentifier.GITHUB webhook_endpoint: Any = GitHubIntegrationsWebhookEndpoint """Overridden in GithubEnterpriseRequestParser""" - always_bucket = True def _get_external_id(self, event: Mapping[str, Any]) -> str | None: """Overridden in GithubEnterpriseRequestParser""" diff --git a/src/sentry/middleware/integrations/parsers/jira.py b/src/sentry/middleware/integrations/parsers/jira.py index d1d09dd0c08c..d0d33bb398a0 100644 --- a/src/sentry/middleware/integrations/parsers/jira.py +++ b/src/sentry/middleware/integrations/parsers/jira.py @@ -37,10 +37,6 @@ class JiraRequestParser(BaseRequestParser): provider = IntegrationProviderSlug.JIRA.value webhook_identifier = WebhookProviderIdentifier.JIRA - # Far lower volume than GitHub: enough to unserialize a burst without thinning - # mailboxes into scheduler rows that each carry a handful of payloads. - mailbox_bucket_count = 10 - control_classes = [ JiraDescriptorEndpoint, JiraSentryInstallationView, diff --git a/src/sentry/middleware/integrations/parsers/vsts.py b/src/sentry/middleware/integrations/parsers/vsts.py index c1d5c5057805..52965febbcbd 100644 --- a/src/sentry/middleware/integrations/parsers/vsts.py +++ b/src/sentry/middleware/integrations/parsers/vsts.py @@ -23,10 +23,6 @@ class VstsRequestParser(BaseRequestParser): provider = IntegrationProviderSlug.AZURE_DEVOPS.value webhook_identifier = WebhookProviderIdentifier.VSTS - # Far lower volume than GitHub: enough to unserialize a burst without thinning - # mailboxes into scheduler rows that each carry a handful of payloads. - mailbox_bucket_count = 10 - cell_view_classes = [WorkItemWebhook] @control_silo_function diff --git a/src/sentry/options/defaults.py b/src/sentry/options/defaults.py index c7e04ad50031..06fc8dbb92ba 100644 --- a/src/sentry/options/defaults.py +++ b/src/sentry/options/defaults.py @@ -2619,6 +2619,26 @@ default=4, flags=FLAG_AUTOMATOR_MODIFIABLE, ) +# How many payloads over the rate window one delivery thread should be worth. Times +# `worker_threads`, this is the depth a mailbox reaches before its split widens, so +# lowering it splits sooner and wider. Tunable because the right value is not known: +# the `buckets` tag on `hybridcloud.webhookpayload.mailbox_routing` is what would +# settle it. +register( + "hybridcloud.webhookpayload.payloads_per_thread", + default=4, + flags=FLAG_AUTOMATOR_MODIFIABLE, +) +# Most mailboxes one integration's split may occupy; past it they simply grow deeper. +# A safety valve on how many scheduler rows and dispatch slots one sender can take. +# Rounded down to a power of two when read: the split climbs a ladder of doublings, +# and a cap off that ladder makes a resize into it re-map nearly every key instead of +# half. +register( + "hybridcloud.webhookpayload.max_mailbox_buckets", + default=64, + flags=FLAG_AUTOMATOR_MODIFIABLE, +) # Remove the rows a claim-bounded drain finishes with — delivered, attempts # exhausted, or stale — in batches instead of one DELETE per row. Such a drain # stays inside a claim reserved for its whole run, so deferring deletes cannot diff --git a/src/sentry/testutils/outbox.py b/src/sentry/testutils/outbox.py index 376c640e4acc..93b74209abd3 100644 --- a/src/sentry/testutils/outbox.py +++ b/src/sentry/testutils/outbox.py @@ -2,7 +2,9 @@ import contextlib import functools +from collections.abc import Generator from typing import Any +from unittest import mock from django.conf import settings from django.core.handlers.wsgi import WSGIRequest @@ -64,6 +66,21 @@ def wrapper(*args: Any, **kwargs: Any) -> Any: raise OutboxRecursionLimitError +@contextlib.contextmanager +def override_mailbox_bucket_count(count: int) -> Generator[None]: + """Pin how wide a parser splits an integration's mailbox. + + The split is sized from the integration's recent webhook rate, so a test that + asserts on a bucket number would otherwise have to send enough webhooks to earn + one first. + """ + with mock.patch( + "sentry.integrations.middleware.hybrid_cloud.parser.mailbox_bucket_count", + return_value=count, + ): + yield + + def assert_no_webhook_payloads() -> None: messages = WebhookPayload.objects.filter().count() assert messages == 0, "No webhookpayload messages should be created" diff --git a/tests/sentry/hybridcloud/test_webhook_mailbox_sizing.py b/tests/sentry/hybridcloud/test_webhook_mailbox_sizing.py new file mode 100644 index 000000000000..faa2113e6024 --- /dev/null +++ b/tests/sentry/hybridcloud/test_webhook_mailbox_sizing.py @@ -0,0 +1,233 @@ +from dataclasses import replace +from time import time +from typing import Any +from unittest.mock import MagicMock, patch + +from django.conf import settings +from redis.exceptions import RedisError + +from sentry.hybridcloud.mailbox import MailboxName +from sentry.hybridcloud.webhook_mailbox_sizing import ( + SHARD_COUNT, + SHARD_SECONDS, + SHARD_TTL_SECONDS, + STRICT_BUCKET_COUNT, + WINDOW_SECONDS, + _count_for_payloads, + _max_buckets, + _rate_counter_key, + _shard_key, + mailbox_bucket_count, +) +from sentry.testutils.cases import TestCase +from sentry.testutils.helpers import override_options +from sentry.testutils.helpers.datetime import freeze_time +from sentry.utils import redis + +THREADS = 4 +PER_THREAD = 4 +DEPTH = THREADS * PER_THREAD + +MAILBOX = MailboxName("github", "4321") +"""A provider the delivery side may reorder, so its width follows its rate.""" + +STRICT_MAILBOX = MailboxName("jira", "4321") +"""A provider delivered in order, so its width is fixed.""" + + +def redis_client() -> Any: + return redis.redis_clusters.get(settings.SENTRY_RATE_LIMIT_REDIS_CLUSTER) + + +def seed_window(mailbox: MailboxName, payloads: int, shards_ago: int = 0) -> None: + """Put `payloads` into one of the mailbox's shards without routing anything.""" + shard = int(time() // SHARD_SECONDS) - shards_ago + counter_key = _rate_counter_key(mailbox) + redis_client().set(_shard_key(counter_key, shard), payloads, ex=SHARD_TTL_SECONDS) + + +class CountForPayloadsTest(TestCase): + def setUp(self) -> None: + super().setUp() + # Pinned rather than left to the option's default: the depth is read per call, + # so a sweep would otherwise put thousands of option lookups behind it. + self.enterContext( + override_options( + { + "hybridcloud.webhookpayload.worker_threads": THREADS, + "hybridcloud.webhookpayload.payloads_per_thread": PER_THREAD, + } + ) + ) + + def test_a_rate_that_cannot_fill_two_mailboxes_does_not_split(self) -> None: + assert _count_for_payloads(0) == 1 + assert _count_for_payloads(DEPTH) == 1 + assert _count_for_payloads(2 * DEPTH - 1) == 1 + + def test_the_rate_has_to_double_to_add_a_bucket(self) -> None: + assert _count_for_payloads(2 * DEPTH) == 2 + assert _count_for_payloads(4 * DEPTH - 1) == 2 + assert _count_for_payloads(4 * DEPTH) == 4 + + def test_every_count_is_a_power_of_two(self) -> None: + """Including the cap: `min` can only preserve the property if it is one too.""" + for rate in range(0, 1_000_000, 37): + count = _count_for_payloads(rate) + assert count & (count - 1) == 0, f"sized {rate} to {count}" + + def test_a_widening_split_moves_half_the_keys(self) -> None: + """Powers of two are what make a resize survivable: a key lands either where + the narrower split put it or exactly that many buckets along.""" + narrow = _count_for_payloads(2 * DEPTH) + wide = _count_for_payloads(4 * DEPTH) + + keys = range(10_000) + unmoved = sum(1 for key in keys if key % narrow == key % wide) + assert unmoved == len(keys) // 2 + + def test_the_split_stops_at_the_cap(self) -> None: + assert _count_for_payloads(10**9) == _max_buckets() + + def test_a_cap_that_is_not_a_power_of_two_is_floored(self) -> None: + """The ladder is doublings, so a cap off it would make a resize into the cap + re-map nearly every key instead of half.""" + with override_options({"hybridcloud.webhookpayload.max_mailbox_buckets": 100}): + assert _count_for_payloads(10**9) == 64 + + def test_an_option_set_to_zero_does_not_divide_by_it(self) -> None: + """Both options are automator-modifiable and neither is validated, and the + drain reads a zero `worker_threads` as one rather than rejecting it.""" + for zeroed in ("payloads_per_thread", "worker_threads"): + with override_options({f"hybridcloud.webhookpayload.{zeroed}": 0}): + assert _count_for_payloads(0) == 1 + assert _count_for_payloads(10**9) == _max_buckets() + + def test_a_rate_redis_could_not_answer_sizes_to_the_cap(self) -> None: + """Wide is the safe direction to be wrong in: an outage must not put a busy + integration back onto one serially drained mailbox.""" + assert _count_for_payloads(None) == _max_buckets() + + +class MailboxBucketCountTest(TestCase): + """Sized against a shallow mailbox so a handful of payloads exercises the split.""" + + def setUp(self) -> None: + super().setUp() + # One delivery thread at four payloads each, so four fill a mailbox. + self.enterContext( + override_options( + { + "hybridcloud.webhookpayload.worker_threads": 1, + "hybridcloud.webhookpayload.payloads_per_thread": 4, + } + ) + ) + + def test_each_payload_is_counted_once(self) -> None: + with freeze_time("2000-01-01"): + for _ in range(3): + mailbox_bucket_count(MAILBOX) + + shard = int(time() // SHARD_SECONDS) + assert redis_client().get(_shard_key(_rate_counter_key(MAILBOX), shard)) == "3" + + def test_the_split_widens_as_the_rate_climbs(self) -> None: + with freeze_time("2000-01-01"): + counts = [mailbox_bucket_count(MAILBOX) for _ in range(32)] + + # Four payloads to a mailbox: one mailbox until the 8th payload of the window, + # two until the 16th, four until the 32nd. + assert counts[:7] == [1] * 7 + assert counts[7:15] == [2] * 8 + assert counts[15:31] == [4] * 16 + assert counts[31] == 8 + + def test_payloads_earlier_in_the_window_still_count(self) -> None: + with freeze_time("2000-01-01"): + seed_window(MAILBOX, payloads=7, shards_ago=SHARD_COUNT - 1) + + assert mailbox_bucket_count(MAILBOX) == 2 + + def test_payloads_older_than_the_window_drop_out(self) -> None: + with freeze_time("2000-01-01"): + seed_window(MAILBOX, payloads=10_000, shards_ago=SHARD_COUNT) + + assert mailbox_bucket_count(MAILBOX) == 1 + + def test_a_shard_expires_without_being_kept_alive(self) -> None: + with freeze_time("2000-01-01"): + mailbox_bucket_count(MAILBOX) + shard = int(time() // SHARD_SECONDS) + + ttl = redis_client().ttl(_shard_key(_rate_counter_key(MAILBOX), shard)) + + # A shard only takes writes while it is the current one, so the fixed TTL + # bounds its life rather than being pushed out by every payload. + assert WINDOW_SECONDS < ttl <= SHARD_TTL_SECONDS + + def test_each_mailbox_family_has_its_own_window(self) -> None: + """A dimension that separates mailboxes ahead of bucketing separates the rate + too, so one busy event type does not size the split for a quiet one.""" + with freeze_time("2000-01-01"): + seed_window(replace(MAILBOX, event_type="push"), payloads=10_000) + + count = mailbox_bucket_count(replace(MAILBOX, event_type="check_run")) + + assert count == 1 + + def test_a_strictly_ordered_provider_has_a_fixed_width(self) -> None: + with freeze_time("2000-01-01"): + seed_window(STRICT_MAILBOX, payloads=10_000) + + assert mailbox_bucket_count(STRICT_MAILBOX) == STRICT_BUCKET_COUNT + + def test_a_strictly_ordered_provider_is_not_counted(self) -> None: + """Nothing sizes from its rate, so nothing should be paying to measure one.""" + with freeze_time("2000-01-01"): + mailbox_bucket_count(STRICT_MAILBOX) + + shard = int(time() // SHARD_SECONDS) + assert redis_client().get(_shard_key(_rate_counter_key(STRICT_MAILBOX), shard)) is None + + def test_a_provider_that_starts_tolerating_reordering_starts_sizing(self) -> None: + """The carve-out dissolves on the option that grants the tolerance, so it + cannot outlive the constraint it exists for.""" + with freeze_time("2000-01-01"): + seed_window(STRICT_MAILBOX, payloads=10_000) + + with override_options( + {"hybridcloud.webhookpayload.skip_on_failure_providers": ["jira"]} + ): + assert mailbox_bucket_count(STRICT_MAILBOX) == _max_buckets() + + def test_a_redis_error_sizes_to_the_cap(self) -> None: + with patch( + "sentry.hybridcloud.webhook_mailbox_sizing.redis.redis_clusters.get", + side_effect=RedisError("unreachable"), + ): + count = mailbox_bucket_count(MAILBOX) + + assert count == _max_buckets() + + def test_a_reply_that_does_not_destructure_sizes_to_the_cap(self) -> None: + """Sizing runs before the payload row is written, so a reply we cannot read has + to fail the same way an outage does rather than 500 a webhook we could queue.""" + pipeline = MagicMock() + pipeline.execute.return_value = [1] + + with patch( + "sentry.hybridcloud.webhook_mailbox_sizing.redis.redis_clusters.get", + return_value=MagicMock(pipeline=MagicMock(return_value=pipeline)), + ): + assert mailbox_bucket_count(MAILBOX) == _max_buckets() + + def test_a_reply_that_does_not_coerce_sizes_to_the_cap(self) -> None: + pipeline = MagicMock() + pipeline.execute.return_value = ["not-a-number", True, []] + + with patch( + "sentry.hybridcloud.webhook_mailbox_sizing.redis.redis_clusters.get", + return_value=MagicMock(pipeline=MagicMock(return_value=pipeline)), + ): + assert mailbox_bucket_count(MAILBOX) == _max_buckets() diff --git a/tests/sentry/integrations/middleware/hybrid_cloud/test_base.py b/tests/sentry/integrations/middleware/hybrid_cloud/test_base.py index f55681c1bde9..dd6c540e8b75 100644 --- a/tests/sentry/integrations/middleware/hybrid_cloud/test_base.py +++ b/tests/sentry/integrations/middleware/hybrid_cloud/test_base.py @@ -3,7 +3,6 @@ from unittest.mock import MagicMock, patch import pytest -from django.core.cache import cache from django.http import HttpResponse from django.test import RequestFactory, override_settings from pytest import raises @@ -19,10 +18,12 @@ from sentry.integrations.middleware.metrics import MiddlewareHaltReason from sentry.integrations.models.integration import Integration from sentry.integrations.models.organization_integration import OrganizationIntegration +from sentry.integrations.types import IntegrationProviderSlug from sentry.silo.base import SiloLimit, SiloMode from sentry.testutils.asserts import assert_failure_metric, assert_halt_metric from sentry.testutils.cases import TestCase from sentry.testutils.helpers.options import override_options +from sentry.testutils.outbox import override_mailbox_bucket_count from sentry.types.cell import Cell @@ -163,7 +164,7 @@ class MockParser(BaseRequestParser): (payload.cell_name, payload.mailbox_name) for payload in WebhookPayload.objects.all() } == {("us", "slack:us:12345"), ("eu", "slack:eu:12345")} - def test_get_mailbox_buckets_only_above_volume(self) -> None: + def test_get_mailbox_buckets_whenever_the_split_is_wide(self) -> None: class BucketedParser(ExampleRequestParser): def mailbox_bucket_id(self, data: dict[str, Any]) -> int | None: return 177 @@ -173,34 +174,13 @@ def mailbox_bucket_id(self, data: dict[str, Any]) -> int | None: ) parser = BucketedParser(self.request, self.response_handler) - with patch( - "sentry.integrations.middleware.hybrid_cloud.parser.ratelimiter.is_limited", - return_value=False, - ): - assert str(parser.get_mailbox(integration, {})) == f"test_provider:{integration.id}" - with patch( - "sentry.integrations.middleware.hybrid_cloud.parser.ratelimiter.is_limited", - return_value=True, - ): - assert str(parser.get_mailbox(integration, {})) == f"test_provider:{integration.id}:77" - - def test_get_mailbox_always_bucket_skips_volume_check(self) -> None: - class AlwaysBucketedParser(ExampleRequestParser): - always_bucket = True - - def mailbox_bucket_id(self, data: dict[str, Any]) -> int | None: - return 177 - - integration = self.create_integration( - organization=self.organization, external_id="1", provider="test_provider" - ) - parser = AlwaysBucketedParser(self.request, self.response_handler) + with override_mailbox_bucket_count(16): + assert str(parser.get_mailbox(integration, {})) == f"test_provider:{integration.id}:1" - with patch( - "sentry.integrations.middleware.hybrid_cloud.parser.ratelimiter.is_limited" - ) as mock_is_limited: - assert str(parser.get_mailbox(integration, {})) == f"test_provider:{integration.id}:77" - mock_is_limited.assert_not_called() + # A split one mailbox wide is the integration mailbox under another name, so + # it keeps the name the mailbox already had. + with override_mailbox_bucket_count(1): + assert str(parser.get_mailbox(integration, {})) == f"test_provider:{integration.id}" @override_settings(SILO_MODE=SiloMode.CONTROL) @patch("sentry.integrations.middleware.hybrid_cloud.parser.maybe_trigger_drain") @@ -305,6 +285,28 @@ def test_shed_inbound_by_integration(self, mock_trigger: MagicMock) -> None: assert WebhookPayload.objects.count() == 2 assert mock_trigger.call_count == 2 + @override_settings(SILO_MODE=SiloMode.CONTROL) + @patch("sentry.integrations.middleware.hybrid_cloud.parser.mailbox_bucket_count") + def test_a_shed_payload_is_left_out_of_the_rate(self, mock_count: MagicMock) -> None: + """Callers build the mailbox as an argument, so it is built before the shed + check runs. A shed payload is never queued, so it must not size the split.""" + integration = self.create_integration( + organization=self.organization, provider="test_provider", external_id="1" + ) + parser = BucketingRequestParser(self.request, self.response_handler) + + with override_options( + { + SHED_INBOUND_KILLSWITCH: [ + {"provider": "test_provider", "integration_id": str(integration.id)} + ] + } + ): + mailbox = parser.get_mailbox(integration, {"bucket_id": 101}) + + assert str(mailbox) == f"test_provider:{integration.id}" + assert not mock_count.called + @override_settings(SILO_MODE=SiloMode.CONTROL) @override_options({SHED_INBOUND_KILLSWITCH: [{"unknown_field": "test_provider"}]}) def test_shed_inbound_ignores_unknown_condition_fields(self) -> None: @@ -409,53 +411,89 @@ def test_get_organizations_from_integration_missing_org_integration( @override_settings(SILO_MODE=SiloMode.CONTROL) @patch("sentry.integrations.middleware.hybrid_cloud.parser.metrics.incr") - def test_mailbox_under_volume_gate(self, mock_incr: MagicMock) -> None: + def test_mailbox_identifier_without_a_bucket_key(self, mock_incr: MagicMock) -> None: integration = self.create_integration( organization=self.organization, provider="test_provider", external_id="test_external_id" ) parser = BucketingRequestParser(self.request, self.response_handler) - assert ( - str(parser.get_mailbox(integration, {"bucket_id": 101})) - == f"test_provider:{integration.id}" - ) + assert str(parser.get_mailbox(integration, {})) == f"test_provider:{integration.id}" mock_incr.assert_any_call( "hybridcloud.webhookpayload.mailbox_routing", - tags={"provider": "test_provider", "bucketed": "false", "reason": "under_volume_gate"}, + tags={"provider": "test_provider", "bucketed": "false", "reason": "no_bucket_key"}, ) @override_settings(SILO_MODE=SiloMode.CONTROL) @patch("sentry.integrations.middleware.hybrid_cloud.parser.metrics.incr") - def test_mailbox_identifier_without_a_bucket_key(self, mock_incr: MagicMock) -> None: + def test_mailbox_identifier_bucketed(self, mock_incr: MagicMock) -> None: integration = self.create_integration( organization=self.organization, provider="test_provider", external_id="test_external_id" ) - cache.set(f"webhookpayload:test_provider:{integration.id}:use_buckets", 1) parser = BucketingRequestParser(self.request, self.response_handler) - assert str(parser.get_mailbox(integration, {})) == f"test_provider:{integration.id}" + with override_mailbox_bucket_count(16): + assert ( + str(parser.get_mailbox(integration, {"bucket_id": 101})) + == f"test_provider:{integration.id}:5" + ) mock_incr.assert_any_call( "hybridcloud.webhookpayload.mailbox_routing", - tags={"provider": "test_provider", "bucketed": "false", "reason": "no_bucket_key"}, + tags={ + "provider": "test_provider", + "bucketed": "true", + "reason": "bucketed", + "buckets": "16", + }, + ) + + @override_settings(SILO_MODE=SiloMode.CONTROL) + @patch( + "sentry.integrations.middleware.hybrid_cloud.parser.mailbox_bucket_count", + return_value=16, + ) + def test_bucketing_is_sized_per_event_type(self, mock_bucket_count: MagicMock) -> None: + class GithubLikeParser(BucketingRequestParser): + provider = IntegrationProviderSlug.GITHUB.value + + def mailbox_event_type(self, data: dict[str, Any]) -> str | None: + return data.get("event_type") + + integration = self.create_integration( + organization=self.organization, provider="github", external_id="github:1" ) + parser = GithubLikeParser(self.request, self.response_handler) + + parser.get_mailbox(integration, {"bucket_id": 101, "event_type": "push"}) + assert mock_bucket_count.call_args.args[0].event_type == "push" + + # An event type the registry does not know never reaches the mailbox name, so + # it must not reach the counter key either: the body is unverified here, and a + # key taken from it verbatim is unbounded Redis keys. + parser.get_mailbox(integration, {"bucket_id": 101, "event_type": "../evil"}) + assert mock_bucket_count.call_args.args[0].event_type is None @override_settings(SILO_MODE=SiloMode.CONTROL) @patch("sentry.integrations.middleware.hybrid_cloud.parser.metrics.incr") - def test_mailbox_identifier_bucketed(self, mock_incr: MagicMock) -> None: + def test_mailbox_identifier_under_the_rate_a_split_needs(self, mock_incr: MagicMock) -> None: integration = self.create_integration( organization=self.organization, provider="test_provider", external_id="test_external_id" ) - cache.set(f"webhookpayload:test_provider:{integration.id}:use_buckets", 1) parser = BucketingRequestParser(self.request, self.response_handler) - assert ( - str(parser.get_mailbox(integration, {"bucket_id": 101})) - == f"test_provider:{integration.id}:1" - ) + with override_mailbox_bucket_count(1): + assert ( + str(parser.get_mailbox(integration, {"bucket_id": 101})) + == f"test_provider:{integration.id}" + ) mock_incr.assert_any_call( "hybridcloud.webhookpayload.mailbox_routing", - tags={"provider": "test_provider", "bucketed": "true", "reason": "bucketed"}, + tags={ + "provider": "test_provider", + "bucketed": "false", + "reason": "under_rate", + "buckets": "1", + }, ) diff --git a/tests/sentry/middleware/integrations/parsers/test_github.py b/tests/sentry/middleware/integrations/parsers/test_github.py index 3f2adb869669..f71b7b7bd729 100644 --- a/tests/sentry/middleware/integrations/parsers/test_github.py +++ b/tests/sentry/middleware/integrations/parsers/test_github.py @@ -21,7 +21,11 @@ from sentry.testutils.cases import TestCase from sentry.testutils.cell import override_cells from sentry.testutils.helpers.options import override_options -from sentry.testutils.outbox import assert_no_webhook_payloads, assert_webhook_payloads_for_mailbox +from sentry.testutils.outbox import ( + assert_no_webhook_payloads, + assert_webhook_payloads_for_mailbox, + override_mailbox_bucket_count, +) from sentry.testutils.silo import control_silo_test from sentry.types.cell import Cell @@ -34,6 +38,12 @@ class GithubRequestParserTest(TestCase): factory = RequestFactory() path = reverse("sentry-integration-github-webhook") + def setUp(self) -> None: + super().setUp() + # One request never sends fast enough to earn a split. Pin the width so these + # assertions stay about which bucket a key lands in; repository 123 lands in 59. + self.enterContext(override_mailbox_bucket_count(64)) + def get_response(self, req: HttpRequest) -> HttpResponse: return HttpResponse(status=200, content="passthrough") @@ -309,6 +319,12 @@ class GithubRequestParserMailboxBucketingTest(TestCase): factory = RequestFactory() path = reverse("sentry-integration-github-webhook") + def setUp(self) -> None: + super().setUp() + # One request never sends fast enough to earn a split. Pin the width so these + # assertions stay about which bucket a key lands in. + self.enterContext(override_mailbox_bucket_count(64)) + def get_response(self, req: HttpRequest) -> HttpResponse: return HttpResponse(status=200, content="passthrough") @@ -368,10 +384,10 @@ def test_webhook_outbox_creation_with_bucketing(self) -> None: assert isinstance(response, HttpResponse) assert response.status_code == status.HTTP_202_ACCEPTED - # 35129377 % 100 = 77, event type appended for per-event-type isolation + # 35129377 % 64 = 33, event type appended for per-event-type isolation assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"github:{integration.id}:77:push", + mailbox_name=f"github:{integration.id}:33:push", cell_names=[cell.name], ) @@ -421,7 +437,7 @@ def test_webhook_outbox_creation_with_bucketing_no_event_type_header(self) -> No # No event type header — identifier is repo-bucket only assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"github:{integration.id}:77", + mailbox_name=f"github:{integration.id}:33", cell_names=[cell.name], ) @@ -457,6 +473,12 @@ class GithubRequestParserDropUnprocessedEventsTest(TestCase): factory = RequestFactory() path = reverse("sentry-integration-github-webhook") + def setUp(self) -> None: + super().setUp() + # One request never sends fast enough to earn a split. Pin the width so these + # assertions stay about which bucket a key lands in. + self.enterContext(override_mailbox_bucket_count(64)) + def get_response(self, req: HttpRequest) -> HttpResponse: return HttpResponse(status=200, content="passthrough") @@ -510,7 +532,7 @@ def test_supported_event_never_dropped(self) -> None: assert response.status_code == status.HTTP_202_ACCEPTED assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"github:{integration.id}:23:push", + mailbox_name=f"github:{integration.id}:59:push", cell_names=[cell.name], ) @@ -533,7 +555,7 @@ def test_missing_x_github_event_forwards_to_cell(self) -> None: assert response.status_code == status.HTTP_202_ACCEPTED assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"github:{integration.id}:23", + mailbox_name=f"github:{integration.id}:59", cell_names=[cell.name], ) @@ -593,7 +615,7 @@ def test_forwards_check_run_completed(self) -> None: assert response.status_code == status.HTTP_202_ACCEPTED assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"github:{integration.id}:23:check_run", + mailbox_name=f"github:{integration.id}:59:check_run", cell_names=[cell.name], ) @@ -610,7 +632,7 @@ def test_forwards_check_run_rerequested(self) -> None: assert response.status_code == status.HTTP_202_ACCEPTED assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"github:{integration.id}:23:check_run", + mailbox_name=f"github:{integration.id}:59:check_run", cell_names=[cell.name], ) @@ -627,7 +649,7 @@ def test_forwards_check_run_requested_action(self) -> None: assert response.status_code == status.HTTP_202_ACCEPTED assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"github:{integration.id}:23:check_run", + mailbox_name=f"github:{integration.id}:59:check_run", cell_names=[cell.name], ) @@ -775,7 +797,7 @@ def test_forwards_check_suite_completed(self) -> None: assert response.status_code == status.HTTP_202_ACCEPTED assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"github:{integration.id}:23:check_suite", + mailbox_name=f"github:{integration.id}:59:check_suite", cell_names=[cell.name], ) diff --git a/tests/sentry/middleware/integrations/parsers/test_gitlab.py b/tests/sentry/middleware/integrations/parsers/test_gitlab.py index 1f81909aa234..e7ea4ed4a599 100644 --- a/tests/sentry/middleware/integrations/parsers/test_gitlab.py +++ b/tests/sentry/middleware/integrations/parsers/test_gitlab.py @@ -1,5 +1,3 @@ -from unittest import mock - import responses from django.db import connections, router, transaction from django.http import HttpRequest, HttpResponse @@ -21,7 +19,11 @@ from sentry.testutils.cases import TestCase from sentry.testutils.cell import override_cells from sentry.testutils.helpers.options import override_options -from sentry.testutils.outbox import assert_no_webhook_payloads, assert_webhook_payloads_for_mailbox +from sentry.testutils.outbox import ( + assert_no_webhook_payloads, + assert_webhook_payloads_for_mailbox, + override_mailbox_bucket_count, +) from sentry.testutils.silo import control_silo_test from sentry.types.cell import Cell @@ -34,6 +36,12 @@ class GitlabRequestParserTest(TestCase): factory = RequestFactory() path = f"{IntegrationClassification.integration_prefix}gitlab/webhook/" + def setUp(self) -> None: + super().setUp() + # One request never sends fast enough to earn a split. Pin the width so these + # assertions stay about which bucket a key lands in. + self.enterContext(override_mailbox_bucket_count(64)) + def get_response(self, req: HttpRequest) -> HttpResponse: return HttpResponse(status=200, content="passthrough") @@ -183,7 +191,7 @@ def test_routing_webhook_properly_with_cells(self) -> None: assert len(responses.calls) == 0 assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"gitlab:{integration.id}:push", + mailbox_name=f"gitlab:{integration.id}:15:push", cell_names=[cell.name], ) @@ -207,7 +215,7 @@ def test_routing_webhook_ignores_an_unhandled_event_type(self) -> None: # An unvalidated suffix would put an arbitrary body value in the mailbox name. assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"gitlab:{integration.id}", + mailbox_name=f"gitlab:{integration.id}:15", cell_names=[cell.name], ) @@ -252,35 +260,6 @@ def test_routing_webhook_properly_with_multiple_orgs(self) -> None: assert response.status_code == 202 assert response.content == b"" assert len(responses.calls) == 0 - assert_webhook_payloads_for_mailbox( - request=request, - mailbox_name=f"gitlab:{integration.id}:push", - cell_names=[cell.name], - ) - - @override_cells(cell_config) - @override_settings(SILO_MODE=SiloMode.CONTROL) - @responses.activate - def test_routing_webhook_with_mailbox_buckets(self) -> None: - integration = self.get_integration() - request = self.factory.post( - self.path, - data=PUSH_EVENT, - content_type="application/json", - HTTP_X_GITLAB_TOKEN=WEBHOOK_TOKEN, - HTTP_X_GITLAB_EVENT="Push Hook", - ) - with mock.patch( - "sentry.integrations.middleware.hybrid_cloud.parser.ratelimiter.is_limited" - ) as mock_is_limited: - mock_is_limited.return_value = True - parser = GitlabRequestParser(request=request, response_handler=self.get_response) - response = parser.get_response() - - assert isinstance(response, HttpResponse) - assert response.status_code == status.HTTP_202_ACCEPTED - assert response.content == b"" - assert len(responses.calls) == 0 assert_webhook_payloads_for_mailbox( request=request, mailbox_name=f"gitlab:{integration.id}:15:push", @@ -347,6 +326,6 @@ def test_webhook_outbox_creation(self) -> None: assert len(responses.calls) == 0 assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"gitlab:{integration.id}:push", + mailbox_name=f"gitlab:{integration.id}:15:push", cell_names=[cell.name], ) diff --git a/tests/sentry/middleware/integrations/parsers/test_jira.py b/tests/sentry/middleware/integrations/parsers/test_jira.py index 9d91acc9867c..23da06483026 100644 --- a/tests/sentry/middleware/integrations/parsers/test_jira.py +++ b/tests/sentry/middleware/integrations/parsers/test_jira.py @@ -3,7 +3,6 @@ from unittest.mock import patch import responses -from django.core.cache import cache from django.http import HttpRequest, HttpResponse from django.test import RequestFactory, override_settings from rest_framework import status @@ -148,8 +147,6 @@ def test_get_response_routing_to_cell_async(self) -> None: @override_cells(cell_config) def test_get_response_routing_to_cell_async_bucketed(self) -> None: integration = self.get_integration() - use_buckets_key = f"webhookpayload:jira:{integration.id}:use_buckets" - cache.set(use_buckets_key, 1) request = self.factory.post( path=f"{self.path_base}/issue-updated/", data={"issue": {"id": "10425"}}, @@ -161,7 +158,6 @@ def test_get_response_routing_to_cell_async_bucketed(self) -> None: method.return_value = integration response = parser.get_response() - cache.delete(use_buckets_key) assert isinstance(response, HttpResponse) assert response.status_code == status.HTTP_202_ACCEPTED assert_webhook_payloads_for_mailbox( diff --git a/tests/sentry/middleware/integrations/parsers/test_jira_server.py b/tests/sentry/middleware/integrations/parsers/test_jira_server.py index 382f0258d3d5..f95975085fca 100644 --- a/tests/sentry/middleware/integrations/parsers/test_jira_server.py +++ b/tests/sentry/middleware/integrations/parsers/test_jira_server.py @@ -2,7 +2,6 @@ from unittest import mock import responses -from django.core.cache import cache from django.http import HttpRequest, HttpResponse from django.test import RequestFactory, override_settings from django.urls import reverse @@ -79,7 +78,7 @@ def test_routing_endpoint_with_integration(self) -> None: assert len(responses.calls) == 0 assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"jira_server:{self.integration.id}", + mailbox_name=f"jira_server:{self.integration.id}:1", cell_names=[cell.name], ) @@ -112,7 +111,7 @@ def test_routing_endpoint_with_integration_no_organization_integration(self) -> @override_cells(cell_config) @override_settings(SILO_MODE=SiloMode.CONTROL) @responses.activate - def test_routing_webhook_with_mailbox_buckets_low_volume(self) -> None: + def test_routing_webhook_buckets_on_issue_id(self) -> None: route = reverse("sentry-extensions-jiraserver-issue-updated", kwargs={"token": "TOKEN"}) request = self.factory.post( @@ -131,71 +130,6 @@ def test_routing_webhook_with_mailbox_buckets_low_volume(self) -> None: assert len(responses.calls) == 0 assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"jira_server:{self.integration.id}", - cell_names=[cell.name], - ) - - @override_cells(cell_config) - @override_settings(SILO_MODE=SiloMode.CONTROL) - @responses.activate - def test_routing_webhook_with_mailbox_buckets_high_volume(self) -> None: - route = reverse("sentry-extensions-jiraserver-issue-updated", kwargs={"token": "TOKEN"}) - - request = self.factory.post( - route, data=issue_updated_payload, content_type="application/json" - ) - parser = JiraServerRequestParser(request=request, response_handler=self.get_response) - - with ( - mock.patch( - "sentry.integrations.middleware.hybrid_cloud.parser.ratelimiter.is_limited" - ) as mock_is_limited, - mock.patch( - "sentry.middleware.integrations.parsers.jira_server.get_integration_from_token" - ) as mock_get_token, - ): - mock_is_limited.return_value = True - mock_get_token.return_value = self.integration - response = parser.get_response() - assert isinstance(response, HttpResponse) - assert response.status_code == status.HTTP_202_ACCEPTED - assert response.content == b"" - assert len(responses.calls) == 0 - assert_webhook_payloads_for_mailbox( - request=request, - # Mailbox name should have an extra segment - mailbox_name=f"jira_server:{self.integration.id}:1", - cell_names=[cell.name], - ) - - @override_cells(cell_config) - @override_settings(SILO_MODE=SiloMode.CONTROL) - @responses.activate - def test_routing_webhook_with_mailbox_bucket_mode_active(self) -> None: - route = reverse("sentry-extensions-jiraserver-issue-updated", kwargs={"token": "TOKEN"}) - - request = self.factory.post( - route, data=issue_updated_payload, content_type="application/json" - ) - parser = JiraServerRequestParser(request=request, response_handler=self.get_response) - - use_bucket_key = f"webhookpayload:jira_server:{self.integration.id}:use_buckets" - cache.set(use_bucket_key, 1) - - with mock.patch( - "sentry.middleware.integrations.parsers.jira_server.get_integration_from_token" - ) as mock_get_token: - mock_get_token.return_value = self.integration - response = parser.get_response() - - cache.delete(use_bucket_key) - assert isinstance(response, HttpResponse) - assert response.status_code == status.HTTP_202_ACCEPTED - assert response.content == b"" - assert len(responses.calls) == 0 - assert_webhook_payloads_for_mailbox( - request=request, - # Mailbox name should have an extra segment mailbox_name=f"jira_server:{self.integration.id}:1", cell_names=[cell.name], ) diff --git a/tests/sentry/middleware/integrations/parsers/test_vsts.py b/tests/sentry/middleware/integrations/parsers/test_vsts.py index 7b858b86e06b..1a08a78a6543 100644 --- a/tests/sentry/middleware/integrations/parsers/test_vsts.py +++ b/tests/sentry/middleware/integrations/parsers/test_vsts.py @@ -1,7 +1,6 @@ from copy import deepcopy import responses -from django.core.cache import cache from django.http import HttpRequest, HttpResponse from django.test import RequestFactory from django.urls import reverse @@ -71,7 +70,7 @@ def test_routing_work_item_webhook(self) -> None: assert response.status_code == 202 assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"vsts:{self.integration.id}", + mailbox_name=f"vsts:{self.integration.id}:1", cell_names=["us"], ) @@ -139,16 +138,16 @@ def test_webhook_outbox_creation(self) -> None: parser.get_response() assert_webhook_payloads_for_mailbox( request=request, - mailbox_name=f"vsts:{self.integration.id}", + mailbox_name=f"vsts:{self.integration.id}:1", cell_names=["us"], ) - def test_webhook_outbox_creation_bucketed(self) -> None: - use_buckets_key = f"webhookpayload:vsts:{self.integration.id}:use_buckets" - cache.set(use_buckets_key, 1) + def test_webhook_outbox_creation_without_a_work_item(self) -> None: + data = deepcopy(WORK_ITEM_UPDATED) + del data["resource"]["workItemId"] request = self.factory.post( self.path, - data=WORK_ITEM_UPDATED, + data=data, content_type="application/json", HTTP_SHARED_SECRET=self.shared_secret, ) @@ -157,11 +156,9 @@ def test_webhook_outbox_creation_bucketed(self) -> None: assert_no_webhook_payloads() parser.get_response() - cache.delete(use_buckets_key) assert_webhook_payloads_for_mailbox( request=request, - # workItemId 31 % 10 - mailbox_name=f"vsts:{self.integration.id}:1", + mailbox_name=f"vsts:{self.integration.id}", cell_names=["us"], )