From 487d42ff9707f2ba4903c6c6d62d3139088b1a6c Mon Sep 17 00:00:00 2001 From: Simon Hellmayr Date: Thu, 10 Sep 2026 13:59:10 +0200 Subject: [PATCH 1/3] ref(dynamic-sampling): Run and serve the per-org pipeline by default The per-org dynamic sampling pipeline only ran, and rules only read its caches, for organizations selected by dynamic-sampling.per_org.rollout-rate and dynamic-sampling.per_org.serving-rollout-rate. Both default to 0.0, so an install without the options automator, such as self-hosted, never ran it and served every project at the fallback rate. Default both rates to 1.0, which is what SaaS already sets, so the EAP-based pipeline works out of the box. Tests that relied on a missing per-org rate falling back to the blended rate now store the rate the per-org pass would have written; the legacy pipeline tests pin the serving rate to 0.0 until they are deleted. Co-Authored-By: Claude Fable 5.1 --- src/sentry/options/defaults.py | 25 +++++++++---------- .../dynamic_sampling/per_org/test_helpers.py | 10 ++++++++ .../dynamic_sampling/tasks/test_tasks.py | 7 ++++++ .../dynamic_sampling/test_generate_rules.py | 25 +++++++++++++++---- tests/sentry/relay/test_config.py | 11 +++++--- 5 files changed, 56 insertions(+), 22 deletions(-) diff --git a/src/sentry/options/defaults.py b/src/sentry/options/defaults.py index 9bb9283cb37b..023f13b3e719 100644 --- a/src/sentry/options/defaults.py +++ b/src/sentry/options/defaults.py @@ -2468,28 +2468,27 @@ flags=FLAG_AUTOMATOR_MODIFIABLE, ) -# Deterministic % rollout of the per-org dynamic sampling pipeline, keyed on -# organization id. A value of 0.0 disables the pipeline for every org; 1.0 -# enables it for every org. Intermediate values select a stable hash-based -# subset so toggling the rate up and down does not reshuffle which orgs run. +# Share of organizations the per-org dynamic sampling pipeline runs for, keyed on +# organization id. 1.0 runs it for every org and is the default, so that the pipeline +# works without any option set; 0.0 stops it for every org. Intermediate values select a +# stable hash-based subset, so lowering and raising the rate does not reshuffle which +# orgs run. register( "dynamic-sampling.per_org.rollout-rate", type=Float, - default=0.0, + default=1.0, flags=FLAG_MODIFIABLE_RATE | FLAG_AUTOMATOR_MODIFIABLE, ) -# Deterministic % rollout of serving the per-org pipeline's results, keyed on organization -# id. Above 0.0, rule generation reads the project, transaction and recalibration sample -# rates of the selected orgs from the per-org caches instead of the legacy ones. An org -# only has per-org cache entries once dynamic-sampling.per_org.rollout-rate selects it too. -# An org switches over as a whole: -# until a pass has stored its project sample rates, rule generation serves all of its -# values from the legacy caches, and from then on all of them from the per-org ones. +# Share of organizations whose rules read the project, transaction and recalibration +# sample rates from the per-org pipeline's caches, keyed on organization id. 1.0 serves +# every org from them and is the default; 0.0 serves every org from the legacy caches. An +# org only has per-org cache entries once dynamic-sampling.per_org.rollout-rate selects it +# too, and a project without a stored per-org rate is sampled in full. register( "dynamic-sampling.per_org.serving-rollout-rate", type=Float, - default=0.0, + default=1.0, flags=FLAG_MODIFIABLE_RATE | FLAG_AUTOMATOR_MODIFIABLE, ) diff --git a/tests/sentry/dynamic_sampling/per_org/test_helpers.py b/tests/sentry/dynamic_sampling/per_org/test_helpers.py index 48d8a5fff3ea..88b746b240ee 100644 --- a/tests/sentry/dynamic_sampling/per_org/test_helpers.py +++ b/tests/sentry/dynamic_sampling/per_org/test_helpers.py @@ -5,6 +5,8 @@ from typing import Any from unittest.mock import MagicMock, Mock, patch +from sentry.dynamic_sampling.models.common import RebalancedItem +from sentry.dynamic_sampling.per_org.cache import set_project_sample_rates from sentry.dynamic_sampling.per_org.configuration import ProjectSampleRates from sentry.dynamic_sampling.per_org.queries import ProjectVolume from sentry.dynamic_sampling.per_org.results import DynamicSamplingResults @@ -37,6 +39,14 @@ def patch_configuration(targets: dict[str, Any]) -> Iterator[dict[str, MagicMock } +def store_per_org_project_sample_rate(project: Project, sample_rate: float) -> None: + """Store the rate as if a per-org pass had balanced the project, so that rules serve it.""" + set_project_sample_rates( + project.organization_id, + [RebalancedItem(id=project.id, count=1, new_sample_rate=sample_rate)], + ) + + def make_project_volume(project_id: int, total: int = 100, keep: int = 25) -> ProjectVolume: return ProjectVolume(project_id=project_id, total=total, keep=keep, drop=max(total - keep, 0)) diff --git a/tests/sentry/dynamic_sampling/tasks/test_tasks.py b/tests/sentry/dynamic_sampling/tasks/test_tasks.py index bd6baa351808..1fd97eb7d54f 100644 --- a/tests/sentry/dynamic_sampling/tasks/test_tasks.py +++ b/tests/sentry/dynamic_sampling/tasks/test_tasks.py @@ -42,6 +42,13 @@ class TasksTestCase(BaseMetricsLayerTestCase, TestCase, SnubaTestCase): + def setUp(self) -> None: + super().setUp() + # These tests run the legacy pipeline end to end, so rules must read its caches. + serve_legacy = override_options({"dynamic-sampling.per_org.serving-rollout-rate": 0.0}) + serve_legacy.__enter__() + self.addCleanup(serve_legacy.__exit__, None, None, None) + @staticmethod def old_date(): return timezone.now() - timedelta(minutes=NEW_MODEL_THRESHOLD_IN_MINUTES + 1) diff --git a/tests/sentry/dynamic_sampling/test_generate_rules.py b/tests/sentry/dynamic_sampling/test_generate_rules.py index aea70b81618c..3212f59d4b36 100644 --- a/tests/sentry/dynamic_sampling/test_generate_rules.py +++ b/tests/sentry/dynamic_sampling/test_generate_rules.py @@ -8,6 +8,7 @@ from sentry.constants import HEALTH_CHECK_GLOBS from sentry.discover.models import TeamKeyTransaction from sentry.dynamic_sampling import ENVIRONMENT_GLOBS, generate_rules, get_redis_client_for_ds +from sentry.dynamic_sampling.per_org.cache import set_adjusted_factor from sentry.dynamic_sampling.rules.base import NEW_MODEL_THRESHOLD_IN_MINUTES from sentry.dynamic_sampling.rules.utils import ( LATEST_RELEASES_BOOST_DECAYED_FACTOR, @@ -21,6 +22,9 @@ from sentry.testutils.helpers import Feature from sentry.testutils.helpers.datetime import freeze_time from sentry.testutils.pytest.fixtures import django_db_all +from tests.sentry.dynamic_sampling.per_org.test_helpers import ( + store_per_org_project_sample_rate, +) @pytest.fixture @@ -133,6 +137,7 @@ def test_generate_rules_return_uniform_rules_with_rate( # it means no enabled user biases get_enabled_user_biases.return_value = {} get_blended_sample_rate.return_value = 0.1 + store_per_org_project_sample_rate(default_old_project, 0.1) assert generate_rules(default_old_project) == [ { "condition": {"inner": [], "op": "and"}, @@ -153,6 +158,7 @@ def test_generate_rules_return_uniform_rules_and_env_rule( get_blended_sample_rate, default_old_project ): get_blended_sample_rate.return_value = 0.1 + store_per_org_project_sample_rate(default_old_project, 0.1) default_old_project.update_option( "sentry:dynamic_sampling_biases", [ @@ -250,6 +256,7 @@ def test_generate_rules_with_different_project_platforms( default_old_project = _apply_old_date_to_project_and_org(default_project) get_blended_sample_rate.return_value = 0.1 + store_per_org_project_sample_rate(default_old_project, 0.1) apply_dynamic_factor.return_value = LATEST_RELEASES_BOOST_FACTOR redis_client = get_redis_client_for_ds() @@ -306,6 +313,7 @@ def test_generate_rules_return_uniform_rules_and_latest_release_rule( default_old_project = _apply_old_date_to_project_and_org(default_project) get_blended_sample_rate.return_value = 0.1 + store_per_org_project_sample_rate(default_old_project, 0.1) apply_dynamic_factor.return_value = LATEST_RELEASES_BOOST_FACTOR redis_client = get_redis_client_for_ds() @@ -387,6 +395,7 @@ def test_generate_rules_does_not_return_rule_with_deleted_release( default_old_project = _apply_old_date_to_project_and_org(default_project) get_blended_sample_rate.return_value = 0.1 + store_per_org_project_sample_rate(default_old_project, 0.1) apply_dynamic_factor.return_value = LATEST_RELEASES_BOOST_FACTOR redis_client = get_redis_client_for_ds() @@ -483,6 +492,7 @@ def test_generate_rules_with_zero_base_sample_rate( get_blended_sample_rate, default_old_project ) -> None: get_blended_sample_rate.return_value = 0.0 + store_per_org_project_sample_rate(default_old_project, 0.0) assert generate_rules(default_old_project) == [ { @@ -510,6 +520,7 @@ def test_generate_rules_return_uniform_rules_and_low_volume_transactions_rules( t1_rate = 0.7 implicit_rate = 0.037 get_blended_sample_rate.return_value = project_sample_rate + store_per_org_project_sample_rate(default_old_project, project_sample_rate) get_transaction_sample_rates.return_value = ( { "t1": t1_rate, @@ -585,6 +596,7 @@ def test_low_volume_transactions_rules_not_returned_when_inactive( get_transaction_sample_rates, get_blended_sample_rate, default_old_project, default_team ): get_blended_sample_rate.return_value = 0.1 + store_per_org_project_sample_rate(default_old_project, 0.1) get_transaction_sample_rates.return_value = ( { "t1": 0.7, @@ -627,7 +639,7 @@ def test_generate_rules_return_uniform_rules_and_recalibrate_orgs_rule( default_old_project = _apply_old_date_to_project_and_org(default_project) get_blended_sample_rate.return_value = 0.1 - redis_client = get_redis_client_for_ds() + store_per_org_project_sample_rate(default_old_project, 0.1) default_old_project.update_option( "sentry:dynamic_sampling_biases", @@ -642,10 +654,7 @@ def test_generate_rules_return_uniform_rules_and_recalibrate_orgs_rule( ) default_factor = 0.5 - redis_client.set( - f"ds::o:{default_old_project.organization.id}:rate_rebalance_factor2", - default_factor, - ) + set_adjusted_factor(default_old_project.organization.id, default_factor) assert generate_rules(default_old_project) == [ { @@ -670,6 +679,7 @@ def test_generate_rules_return_boost_replay_id( get_blended_sample_rate, default_old_project ) -> None: get_blended_sample_rate.return_value = 0.5 + store_per_org_project_sample_rate(default_old_project, 0.5) default_old_project.update_option( "sentry:dynamic_sampling_biases", [ @@ -713,6 +723,7 @@ def test_generate_rules_return_minimum_sample_rate_when_enabled( get_blended_sample_rate, default_old_project ): get_blended_sample_rate.return_value = 0.3 + store_per_org_project_sample_rate(default_old_project, 0.3) default_old_project.update_option( "sentry:dynamic_sampling_biases", [ @@ -762,6 +773,7 @@ def test_generate_rules_minimum_sample_rate_not_included_when_disabled( get_blended_sample_rate, default_old_project ): get_blended_sample_rate.return_value = 0.3 + store_per_org_project_sample_rate(default_old_project, 0.3) default_old_project.update_option( "sentry:dynamic_sampling_biases", [ @@ -790,6 +802,7 @@ def test_generate_rules_minimum_sample_rate_not_included_by_default( get_blended_sample_rate, default_old_project ): get_blended_sample_rate.return_value = 0.3 + store_per_org_project_sample_rate(default_old_project, 0.3) default_old_project.update_option( "sentry:dynamic_sampling_biases", [ @@ -819,6 +832,7 @@ def test_generate_rules_minimum_sample_rate_correct_order( ): with Feature({"organizations:dynamic-sampling-minimum-sample-rate": True}): get_blended_sample_rate.return_value = 0.4 + store_per_org_project_sample_rate(default_old_project, 0.4) default_old_project.update_option( "sentry:dynamic_sampling_biases", [ @@ -893,6 +907,7 @@ def test_generate_rules_trace_health_checks_feature_enabled( get_blended_sample_rate, default_old_project ): get_blended_sample_rate.return_value = 0.4 + store_per_org_project_sample_rate(default_old_project, 0.4) default_old_project.update_option( "sentry:dynamic_sampling_biases", [ diff --git a/tests/sentry/relay/test_config.py b/tests/sentry/relay/test_config.py index 012f02806f9a..392e1d400ebc 100644 --- a/tests/sentry/relay/test_config.py +++ b/tests/sentry/relay/test_config.py @@ -16,6 +16,7 @@ RuleType, get_redis_client_for_ds, ) +from sentry.dynamic_sampling.per_org.cache import set_adjusted_factor from sentry.dynamic_sampling.rules.base import NEW_MODEL_THRESHOLD_IN_MINUTES from sentry.ingest.inbound_filters import CUSTOM_INBOUND_FILTER_ID_PREFIX from sentry.models.project import Project @@ -28,6 +29,9 @@ from sentry.testutils.pytest.fixtures import InstaSnapshotter, django_db_all from sentry.testutils.silo import cell_silo_test from sentry.utils.safe import get_path +from tests.sentry.dynamic_sampling.per_org.test_helpers import ( + store_per_org_project_sample_rate, +) PII_CONFIG = """ { @@ -316,6 +320,7 @@ def test_project_config_with_all_biases_enabled( old_date = datetime.now(tz=timezone.utc) - timedelta(minutes=NEW_MODEL_THRESHOLD_IN_MINUTES + 1) default_project.organization.date_added = old_date default_project.date_added = old_date + store_per_org_project_sample_rate(default_project, 0.1) # We create a team key transaction. TeamKeyTransaction.objects.create( @@ -347,10 +352,7 @@ def test_project_config_with_all_biases_enabled( # Set factor default_factor = 0.5 - redis_client.set( - f"ds::o:{default_project.organization.id}:rate_rebalance_factor2", - default_factor, - ) + set_adjusted_factor(default_project.organization.id, default_factor) with Feature( { @@ -495,6 +497,7 @@ def test_project_config_with_trace_health_checks_enabled( old_date = datetime.now(tz=timezone.utc) - timedelta(minutes=NEW_MODEL_THRESHOLD_IN_MINUTES + 1) default_project.organization.date_added = old_date default_project.date_added = old_date + store_per_org_project_sample_rate(default_project, 0.1) with Feature( { From 5a04a8c1137bf2629f1324fc4e5a39aa80f6dba6 Mon Sep 17 00:00:00 2001 From: Simon Hellmayr Date: Thu, 10 Sep 2026 14:10:05 +0200 Subject: [PATCH 2/3] test(dynamic-sampling): Delete the legacy pipeline tests instead of pinning them The legacy jobs are killswitched and deleted in a separate PR. Deleting their tests here as well keeps this PR green on its own without pinning the serving rate, and the two identical deletions merge cleanly in either order. Co-Authored-By: Claude Fable 5.1 --- .../dynamic_sampling/tasks/test_tasks.py | 1061 ----------------- 1 file changed, 1061 deletions(-) delete mode 100644 tests/sentry/dynamic_sampling/tasks/test_tasks.py diff --git a/tests/sentry/dynamic_sampling/tasks/test_tasks.py b/tests/sentry/dynamic_sampling/tasks/test_tasks.py deleted file mode 100644 index 1fd97eb7d54f..000000000000 --- a/tests/sentry/dynamic_sampling/tasks/test_tasks.py +++ /dev/null @@ -1,1061 +0,0 @@ -from collections.abc import Callable -from datetime import timedelta -from unittest.mock import MagicMock, patch - -import pytest -from django.utils import timezone - -from sentry.dynamic_sampling import RuleType, generate_rules, get_redis_client_for_ds -from sentry.dynamic_sampling.per_org import cache as per_org_cache -from sentry.dynamic_sampling.rules.base import NEW_MODEL_THRESHOLD_IN_MINUTES -from sentry.dynamic_sampling.rules.biases.recalibration_bias import RecalibrationBias -from sentry.dynamic_sampling.tasks.boost_low_volume_projects import boost_low_volume_projects -from sentry.dynamic_sampling.tasks.boost_low_volume_transactions import ( - boost_low_volume_transactions, -) -from sentry.dynamic_sampling.tasks.helpers.boost_low_volume_projects import ( - generate_boost_low_volume_projects_cache_key, -) -from sentry.dynamic_sampling.tasks.helpers.boost_low_volume_transactions import ( - get_transactions_resampling_rates, -) -from sentry.dynamic_sampling.tasks.helpers.recalibrate_orgs import ( - generate_recalibrate_orgs_cache_key, - generate_recalibrate_projects_cache_key, -) -from sentry.dynamic_sampling.tasks.helpers.sliding_window import ( - generate_sliding_window_org_cache_key, - mark_sliding_window_org_executed, -) -from sentry.dynamic_sampling.tasks.recalibrate_orgs import recalibrate_orgs -from sentry.dynamic_sampling.tasks.sliding_window_org import sliding_window_org -from sentry.dynamic_sampling.types import DynamicSamplingMode -from sentry.snuba.metrics.naming_layer.mri import SpanMRI -from sentry.testutils.cases import BaseMetricsLayerTestCase, SnubaTestCase, TestCase -from sentry.testutils.helpers import with_feature -from sentry.testutils.helpers.datetime import freeze_time -from sentry.testutils.helpers.options import override_options - -MOCK_DATETIME = (timezone.now() - timedelta(days=1)).replace( - hour=0, minute=0, second=0, microsecond=0 -) - - -class TasksTestCase(BaseMetricsLayerTestCase, TestCase, SnubaTestCase): - def setUp(self) -> None: - super().setUp() - # These tests run the legacy pipeline end to end, so rules must read its caches. - serve_legacy = override_options({"dynamic-sampling.per_org.serving-rollout-rate": 0.0}) - serve_legacy.__enter__() - self.addCleanup(serve_legacy.__exit__, None, None, None) - - @staticmethod - def old_date(): - return timezone.now() - timedelta(minutes=NEW_MODEL_THRESHOLD_IN_MINUTES + 1) - - @staticmethod - def disable_all_biases(project): - project.update_option( - "sentry:dynamic_sampling_biases", - [ - {"id": RuleType.BOOST_ENVIRONMENTS_RULE.value, "active": False}, - {"id": RuleType.IGNORE_HEALTH_CHECKS_RULE.value, "active": False}, - {"id": RuleType.BOOST_LATEST_RELEASES_RULE.value, "active": False}, - {"id": RuleType.BOOST_KEY_TRANSACTIONS_RULE.value, "active": False}, - {"id": RuleType.BOOST_LOW_VOLUME_TRANSACTIONS_RULE.value, "active": False}, - {"id": RuleType.BOOST_REPLAY_ID_RULE.value, "active": False}, - ], - ) - - def create_old_organization(self, name): - return self.create_organization(name=name, date_added=self.old_date()) - - def create_old_project(self, name, organization): - return self.create_project(name=name, organization=organization, date_added=self.old_date()) - - def create_project_and_add_metrics(self, name, count, org, tags=None, is_old=True): - if tags is None: - tags = {"transaction": "foo_transaction", "is_segment": "true"} - elif "is_segment" not in tags: - tags = {**tags, "is_segment": "true"} - - if is_old: - proj = self.create_old_project(name=name, organization=org) - else: - proj = self.create_project(name=name, organization=org) - - self.disable_all_biases(project=proj) - - self.store_performance_metric( - name=SpanMRI.COUNT_PER_ROOT_PROJECT.value, - tags=tags, - minutes_before_now=30, - value=count, - project_id=proj.id, - org_id=org.id, - ) - - return proj - - def create_project_without_metrics(self, name, org, is_old=True): - if is_old: - proj = self.create_old_project(name=name, organization=org) - else: - proj = self.create_project(name=name, organization=org) - - self.disable_all_biases(project=proj) - - return proj - - -@freeze_time(MOCK_DATETIME) -class TestBoostLowVolumeProjectsTasks(TasksTestCase): - @property - def now(self): - return MOCK_DATETIME - - @staticmethod - def add_sample_rate_per_project(org_id: int, project_id: int, sample_rate: float): - redis_client = get_redis_client_for_ds() - redis_client.hset( - name=generate_boost_low_volume_projects_cache_key(org_id), - key=str(project_id), - value=sample_rate, - ) - - @staticmethod - def sampling_tier_side_effect(*args, **kwargs): - volume = args[1] - - if volume == 20: - return 100_000, 0.25 - # We want to also hardcode the error case, to test how the system reacts to errors. - elif volume == 0: - return None - - return volume, 1.0 - - @staticmethod - def forecasted_volume_side_effect(*args, **kwargs): - return kwargs["volume"] - - @patch("sentry.quotas.backend.get_blended_sample_rate") - def test_boost_low_volume_projects_with_no_dynamic_sampling(self, get_blended_sample_rate): - get_blended_sample_rate.return_value = 0.25 - test_org = self.create_old_organization(name="sample-org") - - self.create_project_and_add_metrics("a", 9, test_org) - self.create_project_and_add_metrics("b", 7, test_org) - self.create_project_and_add_metrics("c", 3, test_org) - self.create_project_and_add_metrics("d", 1, test_org) - - with self.tasks(): - sliding_window_org() - boost_low_volume_projects() - - @with_feature("organizations:dynamic-sampling") - @patch("sentry.quotas.backend.get_blended_sample_rate") - def test_boost_low_volume_projects_simple( - self, - get_blended_sample_rate, - ): - get_blended_sample_rate.return_value = 0.25 - # Create a org - test_org = self.create_old_organization(name="sample-org") - - # Create 4 projects - proj_a = self.create_project_and_add_metrics("a", 9, test_org) - proj_b = self.create_project_and_add_metrics("b", 7, test_org) - proj_c = self.create_project_and_add_metrics("c", 3, test_org) - proj_d = self.create_project_and_add_metrics("d", 1, test_org) - - with self.tasks(): - sliding_window_org() - boost_low_volume_projects() - - # we expect only uniform rule - # also we test here that `generate_rules` can handle trough redis long floats - assert generate_rules(proj_a)[0]["samplingValue"] == { - "type": "sampleRate", - "value": pytest.approx(0.14814814814814817), - } - assert generate_rules(proj_b)[0]["samplingValue"] == { - "type": "sampleRate", - "value": pytest.approx(0.1904761904761905), - } - assert generate_rules(proj_c)[0]["samplingValue"] == { - "type": "sampleRate", - "value": pytest.approx(0.4444444444444444), - } - assert generate_rules(proj_d)[0]["samplingValue"] == {"type": "sampleRate", "value": 1.0} - - @with_feature("organizations:dynamic-sampling") - @override_options({"dynamic-sampling.legacy.killswitch": True}) - @patch("sentry.quotas.backend.get_blended_sample_rate") - def test_boost_low_volume_projects_killswitch(self, get_blended_sample_rate: MagicMock) -> None: - get_blended_sample_rate.return_value = 0.25 - test_org = self.create_old_organization(name="sample-org") - self.create_project_and_add_metrics("a", 9, test_org) - self.create_project_and_add_metrics("b", 1, test_org) - - with self.tasks(): - boost_low_volume_projects() - - redis_client = get_redis_client_for_ds() - assert redis_client.hgetall(generate_boost_low_volume_projects_cache_key(test_org.id)) == {} - - @with_feature("organizations:dynamic-sampling") - @patch("sentry.quotas.backend.get_blended_sample_rate") - def test_boost_low_volume_projects_simple_with_empty_project( - self, - get_blended_sample_rate, - ): - get_blended_sample_rate.return_value = 0.25 - test_org = self.create_old_organization(name="sample-org") - - proj_a = self.create_project_and_add_metrics("a", 9, test_org) - proj_b = self.create_project_and_add_metrics("b", 7, test_org) - proj_c = self.create_project_and_add_metrics("c", 3, test_org) - proj_d = self.create_project_and_add_metrics("d", 1, test_org) - proj_e = self.create_project_without_metrics("e", test_org) - - with self.tasks(): - sliding_window_org() - boost_low_volume_projects() - - # we expect only uniform rule - # also we test here that `generate_rules` can handle trough redis long floats - assert generate_rules(proj_a)[0]["samplingValue"] == { - "type": "sampleRate", - "value": pytest.approx(0.14814814814814817), - } - assert generate_rules(proj_b)[0]["samplingValue"] == { - "type": "sampleRate", - "value": pytest.approx(0.1904761904761905), - } - assert generate_rules(proj_c)[0]["samplingValue"] == { - "type": "sampleRate", - "value": pytest.approx(0.4444444444444444), - } - assert generate_rules(proj_d)[0]["samplingValue"] == {"type": "sampleRate", "value": 1.0} - assert generate_rules(proj_e)[0]["samplingValue"] == {"type": "sampleRate", "value": 1.0} - - @with_feature("organizations:dynamic-sampling") - @patch("sentry.quotas.backend.get_blended_sample_rate") - @patch("sentry.quotas.backend.get_transaction_sampling_tier_for_volume") - @patch("sentry.dynamic_sampling.tasks.common.extrapolate_monthly_volume") - def test_boost_low_volume_projects_simple_with_sliding_window_org_from_cache( - self, - extrapolate_monthly_volume, - get_transaction_sampling_tier_for_volume, - get_blended_sample_rate, - ): - extrapolate_monthly_volume.side_effect = self.forecasted_volume_side_effect - get_transaction_sampling_tier_for_volume.side_effect = self.sampling_tier_side_effect - get_blended_sample_rate.return_value = 0.8 - - test_org = self.create_old_organization(name="sample-org") - - proj_a = self.create_project_and_add_metrics("a", 9, test_org) - proj_b = self.create_project_and_add_metrics("b", 7, test_org) - proj_c = self.create_project_and_add_metrics("c", 3, test_org) - proj_d = self.create_project_and_add_metrics("d", 1, test_org) - - with self.tasks(): - sliding_window_org() - boost_low_volume_projects() - - # we expect only uniform rule - # also we test here that `generate_rules` can handle trough redis long floats - assert generate_rules(proj_a)[0]["samplingValue"] == { - "type": "sampleRate", - "value": pytest.approx(0.14814814814814817), - } - assert generate_rules(proj_b)[0]["samplingValue"] == { - "type": "sampleRate", - "value": pytest.approx(0.1904761904761905), - } - assert generate_rules(proj_c)[0]["samplingValue"] == { - "type": "sampleRate", - "value": pytest.approx(0.4444444444444444), - } - assert generate_rules(proj_d)[0]["samplingValue"] == {"type": "sampleRate", "value": 1.0} - - @with_feature("organizations:dynamic-sampling") - @patch( - "sentry.dynamic_sampling.tasks.boost_low_volume_projects.schedule_invalidate_project_config" - ) - @patch("sentry.quotas.backend.get_blended_sample_rate") - @patch("sentry.quotas.backend.get_transaction_sampling_tier_for_volume") - @patch("sentry.dynamic_sampling.tasks.common.extrapolate_monthly_volume") - def test_config_invalidation_when_sample_rates_change( - self, - extrapolate_monthly_volume, - get_transaction_sampling_tier_for_volume, - get_blended_sample_rate, - schedule_invalidate_project_config, - ): - extrapolate_monthly_volume.side_effect = self.forecasted_volume_side_effect - get_transaction_sampling_tier_for_volume.side_effect = self.sampling_tier_side_effect - get_blended_sample_rate.return_value = 0.8 - - test_org = self.create_old_organization(name="sample-org") - - proj_a = self.create_project_and_add_metrics("a", 9, test_org) - proj_b = self.create_project_and_add_metrics("b", 7, test_org) - - self.add_sample_rate_per_project(org_id=test_org.id, project_id=proj_a.id, sample_rate=0.1) - self.add_sample_rate_per_project(org_id=test_org.id, project_id=proj_b.id, sample_rate=0.2) - - with self.tasks(): - sliding_window_org() - boost_low_volume_projects() - - assert schedule_invalidate_project_config.call_count == 2 - - @with_feature("organizations:dynamic-sampling") - @patch( - "sentry.dynamic_sampling.tasks.boost_low_volume_projects.schedule_invalidate_project_config" - ) - @patch("sentry.quotas.backend.get_blended_sample_rate") - @patch("sentry.quotas.backend.get_transaction_sampling_tier_for_volume") - @patch("sentry.dynamic_sampling.tasks.common.extrapolate_monthly_volume") - def test_config_invalidation_when_sample_rates_do_not_change( - self, - extrapolate_monthly_volume, - get_transaction_sampling_tier_for_volume, - get_blended_sample_rate, - schedule_invalidate_project_config, - ): - extrapolate_monthly_volume.side_effect = self.forecasted_volume_side_effect - get_transaction_sampling_tier_for_volume.side_effect = self.sampling_tier_side_effect - get_blended_sample_rate.return_value = 1.0 - - test_org = self.create_old_organization(name="sample-org") - - proj_a = self.create_project_and_add_metrics("a", 9, test_org) - proj_b = self.create_project_and_add_metrics("b", 7, test_org) - - self.add_sample_rate_per_project(org_id=test_org.id, project_id=proj_a.id, sample_rate=1.0) - self.add_sample_rate_per_project(org_id=test_org.id, project_id=proj_b.id, sample_rate=1.0) - - with self.tasks(): - boost_low_volume_projects() - - schedule_invalidate_project_config.assert_not_called() - - -@freeze_time(MOCK_DATETIME) -class TestBoostLowVolumeTransactionsTasks(TasksTestCase): - @property - def now(self): - return MOCK_DATETIME - - def setUp(self) -> None: - super().setUp() - self.orgs_info = [] - num_orgs = 3 - num_proj_per_org = 3 - for org_idx in range(num_orgs): - org = self.create_old_organization(f"test-org{org_idx}") - org_info = {"org_id": org.id, "project_ids": []} - self.orgs_info.append(org_info) - for proj_idx in range(num_proj_per_org): - p = self.create_old_project(name=f"test-project-{proj_idx}", organization=org) - org_info["project_ids"].append(p.id) - # create 5 transaction types - for name in ["ts1", "ts2", "tm3", "tl4", "tl5"]: - # make up some unique count - idx = org_idx * num_orgs + proj_idx - num_transactions = self.get_count_for_transaction(idx, name) - self.store_performance_metric( - name=SpanMRI.COUNT_PER_ROOT_PROJECT.value, - tags={"transaction": name, "is_segment": "true"}, - minutes_before_now=30, - value=num_transactions, - project_id=p.id, - org_id=org.id, - ) - self.org_ids = [org["org_id"] for org in self.orgs_info] - - def get_count_for_transaction(self, idx: int, name: str): - """ - Create some known count based on transaction name and the order (based on org and project) - """ - counts = { - "ts1": 1, - "ts2": 100, - "tm3": 1000, - "tl4": 2000, - "tl5": 3000, - } - return idx + counts[name] - - @staticmethod - def flush_redis(): - get_redis_client_for_ds().flushdb() - - @staticmethod - def set_boost_low_volume_projects_cache_entry(org_id: int, project_id: int, value: str): - redis = get_redis_client_for_ds() - cache_key = generate_boost_low_volume_projects_cache_key(org_id=org_id) - redis.hset(name=cache_key, key=str(project_id), value=value) - - def set_boost_low_volume_projects_sample_rate( - self, org_id: int, project_id: int, sample_rate: float - ): - self.set_boost_low_volume_projects_cache_entry(org_id, project_id, str(sample_rate)) - - def set_prioritise_by_project_invalid(self, org_id: int, project_id: int): - # We want also to test for this case in order to verify the fallback to the `get_blended_sample_rate`. - self.set_boost_low_volume_projects_cache_entry(org_id, project_id, "invalid") - - def for_all_orgs_and_projects(self, block: Callable[[int, int], None]): - for org in self.orgs_info: - org_id = org["org_id"] - for project_id in org["project_ids"]: - block(org_id, project_id) - - def set_boost_low_volume_projects_invalid_for_all(self): - self.for_all_orgs_and_projects( - lambda org_id, project_id: self.set_prioritise_by_project_invalid(org_id, project_id) - ) - - def set_boost_low_volume_projects_for_all(self, sample_rate: float): - self.for_all_orgs_and_projects( - lambda org_id, project_id: self.set_boost_low_volume_projects_sample_rate( - org_id, project_id, sample_rate - ) - ) - - @patch("sentry.quotas.backend.get_blended_sample_rate") - def test_boost_low_volume_transactions_with_blended_sample_rate_and_no_dynamic_sampling( - self, get_blended_sample_rate - ): - """ - Create orgs projects & transactions and then check that the rebalancing model is not called because dynamic - sampling is disabled - """ - BLENDED_RATE = 0.25 - get_blended_sample_rate.return_value = BLENDED_RATE - - with self.tasks(): - boost_low_volume_transactions() - - @with_feature("organizations:dynamic-sampling") - @patch("sentry.quotas.backend.get_blended_sample_rate") - def test_boost_low_volume_transactions_with_blended_sample_rate( - self, get_blended_sample_rate: MagicMock - ) -> None: - """ - Create orgs projects & transactions and then check that the task creates rebalancing data - in Redis. - """ - BLENDED_RATE = 0.25 - get_blended_sample_rate.return_value = BLENDED_RATE - - with self.tasks(): - boost_low_volume_transactions() - - # now redis should contain rebalancing data for our projects - for org in self.orgs_info: - org_id = org["org_id"] - for proj_id in org["project_ids"]: - tran_rate, global_rate = get_transactions_resampling_rates( - org_id=org_id, proj_id=proj_id, default_rate=0.1 - ) - for transaction_name in ["ts1", "ts2", "tm3", "tl4", "tl5"]: - assert ( - transaction_name in tran_rate - ) # check we have some rate calculated for each transaction - assert global_rate == BLENDED_RATE - - @with_feature("organizations:dynamic-sampling") - @override_options({"dynamic-sampling.legacy.killswitch": True}) - @patch("sentry.quotas.backend.get_blended_sample_rate") - def test_boost_low_volume_transactions_killswitch( - self, get_blended_sample_rate: MagicMock - ) -> None: - get_blended_sample_rate.return_value = 0.25 - - with self.tasks(): - boost_low_volume_transactions() - - for org in self.orgs_info: - for proj_id in org["project_ids"]: - tran_rate, _ = get_transactions_resampling_rates( - org_id=org["org_id"], proj_id=proj_id, default_rate=0.1 - ) - assert tran_rate == {} - - @with_feature("organizations:dynamic-sampling") - @patch("sentry.quotas.backend.get_blended_sample_rate") - def test_boost_low_volume_transactions_with_sliding_window_org( - self, get_blended_sample_rate: MagicMock - ) -> None: - """ - Create orgs projects & transactions and then check that the task creates rebalancing data - in Redis with the sliding window per org enabled. - """ - BLENDED_RATE = 0.25 - get_blended_sample_rate.return_value = BLENDED_RATE - - for sliding_window_step, used_sample_rate in ((1, 1.0), (2, BLENDED_RATE), (3, 0.5)): - # We flush redis after each run, to make sure no data persists. - self.flush_redis() - - # No value in cache and sliding window org executed. - if sliding_window_step == 1: - mark_sliding_window_org_executed() - # Invalid value in cache. - elif sliding_window_step == 2: - self.set_boost_low_volume_projects_invalid_for_all() - # Value in cache. - elif sliding_window_step == 3: - self.set_boost_low_volume_projects_for_all(used_sample_rate) - - with self.tasks(): - boost_low_volume_transactions() - - # now redis should contain rebalancing data for our projects - for org in self.orgs_info: - org_id = org["org_id"] - for proj_id in org["project_ids"]: - tran_rate, global_rate = get_transactions_resampling_rates( - org_id=org_id, proj_id=proj_id, default_rate=0.1 - ) - - if sliding_window_step == 1: - # If the sample rate is 100%, we will not find anything in cache, since we don't - # need to run and store the rebalancing. - assert tran_rate == {} - else: - # If the sample rate is < 100%, we want to check that in cache we have a value with - # the correct global rate. - for transaction_name in ["ts1", "ts2", "tm3", "tl4", "tl5"]: - assert ( - transaction_name in tran_rate - ) # check we have some rate calculated for each transaction - - assert global_rate == used_sample_rate - - @with_feature("organizations:dynamic-sampling") - @patch("sentry.quotas.backend.get_blended_sample_rate") - def test_boost_low_volume_transactions_partial( - self, get_blended_sample_rate: MagicMock - ) -> None: - """ - Test the V2 algorithm is used, only specified projects are balanced and the - rest get a global rate - - Create orgs projects & transactions and then check that the task creates rebalancing data - in Redis - """ - BLENDED_RATE = 0.25 - get_blended_sample_rate.return_value = BLENDED_RATE - - with self.options( - { - "dynamic-sampling.prioritise_transactions.num_explicit_large_transactions": 1, - "dynamic-sampling.prioritise_transactions.rebalance_intensity": 0.7, - } - ): - with self.tasks(): - boost_low_volume_transactions() - - # now redis should contain rebalancing data for our projects - for org in self.orgs_info: - org_id = org["org_id"] - for proj_id in org["project_ids"]: - tran_rate, implicit_rate = get_transactions_resampling_rates( - org_id=org_id, proj_id=proj_id, default_rate=0.1 - ) - # explicit transactions - for transaction_name in ["tl5"]: - assert ( - transaction_name in tran_rate - ) # check we have some rate calculated for each transaction - # implicit transactions - for transaction_name in ["ts1", "ts2", "tm3", "tl4"]: - assert ( - transaction_name not in tran_rate - ) # check we have some rate calculated for each transaction - # we do have some different rate for implicit transactions - assert implicit_rate != BLENDED_RATE - - -@freeze_time(MOCK_DATETIME) -class TestRecalibrateOrgsTasks(TasksTestCase): - @property - def now(self): - return MOCK_DATETIME - - def setUp(self) -> None: - super().setUp() - self.orgs_info = [] - self.orgs = [] - self.num_proj = 2 - self.orgs_sampling = [10, 20, 40] - # create some orgs, projects and transactions - for org_rate in self.orgs_sampling: - org = self.create_old_organization(f"test-org-{org_rate}") - org_info = {"org_id": org.id, "project_ids": [], "projects": []} - self.orgs_info.append(org_info) - self.orgs.append(org) - for proj_idx in range(self.num_proj): - p = self.create_old_project(name=f"test-project-{proj_idx}", organization=org) - org_info["projects"].append(p) - org_info["project_ids"].append(p.id) - self.add_metrics(org, p, org_rate) - - def add_metrics(self, org, project, sample_rate): - base_tags = {"transaction": "trans-x", "is_segment": "true"} - - if sample_rate < 100: - self.store_performance_metric( - name=SpanMRI.COUNT_PER_ROOT_PROJECT.value, - tags={**base_tags, "decision": "drop"}, - minutes_before_now=2, - value=100 - sample_rate, - project_id=project.id, - org_id=org.id, - ) - if sample_rate > 0: - self.store_performance_metric( - name=SpanMRI.COUNT_PER_ROOT_PROJECT.value, - tags={**base_tags, "decision": "keep"}, - minutes_before_now=2, - value=sample_rate, - project_id=project.id, - org_id=org.id, - ) - - def add_measure_metrics( - self, - org, - project, - *, - segment_keep: int, - segment_drop: int, - ) -> None: - segment_tags = {"transaction": "trans-x", "is_segment": "true"} - if segment_drop: - self.store_performance_metric( - name=SpanMRI.COUNT_PER_ROOT_PROJECT.value, - tags={**segment_tags, "decision": "drop"}, - minutes_before_now=2, - value=segment_drop, - project_id=project.id, - org_id=org.id, - ) - if segment_keep: - self.store_performance_metric( - name=SpanMRI.COUNT_PER_ROOT_PROJECT.value, - tags={**segment_tags, "decision": "keep"}, - minutes_before_now=2, - value=segment_keep, - project_id=project.id, - org_id=org.id, - ) - - @staticmethod - def set_sliding_window_org_cache_entry(org_id: int, value: str): - redis = get_redis_client_for_ds() - cache_key = generate_sliding_window_org_cache_key(org_id=org_id) - redis.set(cache_key, value) - - def set_sliding_window_org_sample_rate(self, org_id: int, sample_rate: float): - self.set_sliding_window_org_cache_entry(org_id, str(sample_rate)) - - def for_all_orgs(self, block: Callable[[int], None]): - for org in self.orgs_info: - org_id = org["org_id"] - block(org_id) - - def set_sliding_window_org_sample_rate_for_all(self, sample_rate: float): - self.for_all_orgs( - lambda org_id: self.set_sliding_window_org_sample_rate(org_id, sample_rate) - ) - - @patch("sentry.dynamic_sampling.tasks.recalibrate_orgs.compute_adjusted_factor") - @patch("sentry.quotas.backend.get_blended_sample_rate") - def test_recalibrate_orgs_with_no_dynamic_sampling( - self, get_blended_sample_rate, computed_adjusted_factor - ): - """ - Test that the recalibration of orgs doesn't happen if dynamic sampling is not enabled - """ - get_blended_sample_rate.return_value = 0.1 - self.set_sliding_window_org_sample_rate_for_all(0.2) - - with self.tasks(): - recalibrate_orgs() - - computed_adjusted_factor.assert_not_called() - - @with_feature("organizations:dynamic-sampling") - @patch("sentry.quotas.backend.get_blended_sample_rate") - def test_recalibrate_orgs_with_sliding_window_org( - self, get_blended_sample_rate: MagicMock - ) -> None: - """ - Test that the org are going to be rebalanced at 20% and that the sample rate used is the one from the sliding - window org. - - The first org is 10%, so we should increase the sampling - The second org is at 20%, so we are spot on - The third is at 40%, so we should decrease the sampling - """ - get_blended_sample_rate.return_value = 0.1 - self.set_sliding_window_org_sample_rate_for_all(0.2) - - redis_client = get_redis_client_for_ds() - - with self.tasks(): - recalibrate_orgs() - - for idx, org in enumerate(self.orgs): - cache_key = generate_recalibrate_orgs_cache_key(org.id) - val = redis_client.get(cache_key) - - if idx == 0: - assert val is not None - # we sampled at 10% half of what we want so we should adjust by 2 - assert float(val) == 2.0 - elif idx == 1: - # we sampled at 20% we should be spot on (no adjustment) - assert val is None - elif idx == 2: - assert val is not None - # we sampled at 40% twice as much as we wanted we should adjust by 0.5 - assert float(val) == 0.5 - - # now if we run it again (with the same data in the database, the algorithm - # should double down... the previous factor didn't do anything so apply it again) - with self.tasks(): - recalibrate_orgs() - - for idx, org in enumerate(self.orgs): - cache_key = generate_recalibrate_orgs_cache_key(org.id) - val = redis_client.get(cache_key) - - if idx == 0: - assert val is not None - # we sampled at 10% when already having a factor of two half of what we want so we - # should double the current factor to 4 - assert float(val) == 4.0 - elif idx == 1: - # we sampled at 20% we should be spot on (no adjustment) - assert val is None - elif idx == 2: - assert val is not None - # we sampled at 40% twice as much as we wanted we already have a factor of 0.5 - # half it again to 0.25 - assert float(val) == 0.25 - - @with_feature("organizations:dynamic-sampling") - @override_options({"dynamic-sampling.legacy.killswitch": True}) - @patch("sentry.quotas.backend.get_blended_sample_rate") - def test_recalibrate_orgs_killswitch(self, get_blended_sample_rate: MagicMock) -> None: - get_blended_sample_rate.return_value = 0.1 - self.set_sliding_window_org_sample_rate_for_all(0.2) - - with self.tasks(): - recalibrate_orgs() - - redis_client = get_redis_client_for_ds() - for org in self.orgs: - assert redis_client.get(generate_recalibrate_orgs_cache_key(org.id)) is None - - @with_feature("organizations:dynamic-sampling") - @patch("sentry.quotas.backend.get_blended_sample_rate") - def test_recalibrate_orgs_skips_orgs_served_the_per_org_factor( - self, get_blended_sample_rate: MagicMock - ) -> None: - """An org served the per-org factor keeps the legacy cache untouched. - - Both loops step by previous * target / measured. Writing here as well would step a - factor nothing applies, so it would walk to a rebalance bound. - """ - get_blended_sample_rate.return_value = 0.1 - self.set_sliding_window_org_sample_rate_for_all(0.2) - - served_org = self.orgs[0] - redis_client = get_redis_client_for_ds() - - with ( - override_options({"dynamic-sampling.per_org.serving-org-ids": [served_org.id]}), - self.tasks(), - ): - recalibrate_orgs() - - # The served org sampled at 10% against a 20% target, so the legacy task would have - # written a factor of 2.0 here. - assert redis_client.get(generate_recalibrate_orgs_cache_key(served_org.id)) is None - - # Every other org still has its factor written by the legacy task. - other_factor = redis_client.get(generate_recalibrate_orgs_cache_key(self.orgs[2].id)) - assert other_factor is not None - assert float(other_factor) == 0.5 - - @with_feature("organizations:dynamic-sampling") - def test_recalibrate_orgs_continues_from_the_per_org_factor_after_switching_back( - self, - ) -> None: - """An org switched back to the legacy pipeline steps from the factor it was served. - - The legacy key expired while the per-org pipeline served the org, so without the - carry-over the correction would restart from 1.0 and drop the whole boost. - """ - self.set_sliding_window_org_sample_rate_for_all(0.2) - - # This org stored metrics at a 10% sampling rate, so it measures at 0.1. - switched_back_org = self.orgs[0] - per_org_cache.set_adjusted_factor(switched_back_org.id, 3.0) - - redis_client = get_redis_client_for_ds() - - with self.tasks(): - recalibrate_orgs() - - # 3.0 * (0.2 target / 0.1 measured), instead of the 2.0 a restart from 1.0 gives. - factor = redis_client.get(generate_recalibrate_orgs_cache_key(switched_back_org.id)) - assert factor is not None - assert float(factor) == 6.0 - - @with_feature("organizations:dynamic-sampling") - @with_feature("organizations:dynamic-sampling-custom") - def test_recalibrate_orgs_with_custom_ds(self) -> None: - """ - Test several organizations with mixed sampling mode. - - The first org is 10%, so we should increase the sampling - The second org is at 20%, so we are spot on - The third is at 40%, so we should decrease the sampling - """ - - # First two orgs have a 20% sample rate configured, third one is in project mode - self.orgs[0].update_option("sentry:target_sample_rate", 0.2) - self.orgs[1].update_option("sentry:target_sample_rate", 0.2) - self.orgs[2].update_option("sentry:sampling_mode", DynamicSamplingMode.PROJECT) - - # First project gets same 20% sample rate, the other one stays at implicit 100% - p1, p2 = self.orgs_info[2]["projects"] - p1.update_option("sentry:target_sample_rate", 0.2) - - with self.tasks(): - recalibrate_orgs() - - redis_client = get_redis_client_for_ds() - - # First org was sampled at 10%, should be recalibrated at 2x to 20%. - assert redis_client.get(generate_recalibrate_orgs_cache_key(self.orgs[0].id)) == "2.0" - # Second org was sampled at 20%, should not be recalibrated. - assert redis_client.get(generate_recalibrate_orgs_cache_key(self.orgs[1].id)) is None - - # Third org should not have org-level recalibration. - assert redis_client.get(generate_recalibrate_orgs_cache_key(self.orgs[2].id)) is None - # First project was sampled at 40%, should be recalibrated at 0.5x to 20%. - assert redis_client.get(generate_recalibrate_projects_cache_key(p1.id)) == "0.5" - # Second project was sampled at 40%, should be recalibrated at 2.5x to 100%. - assert redis_client.get(generate_recalibrate_projects_cache_key(p2.id)) == "2.5" - - assert RecalibrationBias().generate_rules(p1, base_sample_rate=1.0) == [ - { - "samplingValue": {"type": "factor", "value": 0.5}, - "type": "trace", - "condition": {"op": "and", "inner": []}, - "id": 1004, - } - ] - - assert RecalibrationBias().generate_rules(p2, base_sample_rate=1.0) == [ - { - "samplingValue": {"type": "factor", "value": 2.5}, - "type": "trace", - "condition": {"op": "and", "inner": []}, - "id": 1004, - } - ] - - @with_feature("organizations:dynamic-sampling") - @patch("sentry.quotas.backend.get_blended_sample_rate") - def test_rules_generation_with_recalibrate_orgs( - self, get_blended_sample_rate: MagicMock - ) -> None: - """ - Test that we pass rebalancing values all the way to the rules. - """ - get_blended_sample_rate.return_value = 0.20 - - with self.tasks(): - recalibrate_orgs() - - for org_idx, org in enumerate(self.orgs): - for project in org.project_set.all(): - rules = RecalibrationBias().generate_rules(project, base_sample_rate=0.5) - if org_idx == 0: - # we sampled at 10% half of what we want so we should adjust by 2 - assert rules == [ - { - "samplingValue": {"type": "factor", "value": 2.0}, - "type": "trace", - "condition": {"op": "and", "inner": []}, - "id": 1004, - } - ] - elif org_idx == 1: - # we sampled at 20% we should be spot on (no rule) - assert rules == [] - elif org_idx == 2: - # we sampled at 40% twice as much as we wanted we should adjust by 0.5 - assert rules == [ - { - "samplingValue": {"type": "factor", "value": 0.5}, - "type": "trace", - "condition": {"op": "and", "inner": []}, - "id": 1004, - } - ] - - @with_feature("organizations:dynamic-sampling") - @patch("sentry.quotas.backend.get_blended_sample_rate") - def test_recalibrate_orgs_uses_segments_measure( - self, get_blended_sample_rate: MagicMock - ) -> None: - """ - Test that all orgs use segment metrics (SEGMENTS measure) for recalibration. - """ - get_blended_sample_rate.return_value = 0.1 - self.set_sliding_window_org_sample_rate_for_all(0.2) - - redis_client = get_redis_client_for_ds() - - with self.tasks(): - recalibrate_orgs() - - # First org should be recalibrated (sampled at 10%, target 20% -> factor 2.0) - cache_key = generate_recalibrate_orgs_cache_key(self.orgs[0].id) - val = redis_client.get(cache_key) - assert val is not None - assert float(val) == 2.0 - - # Second org sampled at 20%, target 20% -> no adjustment needed - cache_key = generate_recalibrate_orgs_cache_key(self.orgs[1].id) - val = redis_client.get(cache_key) - assert val is None - - # Third org sampled at 40%, target 20% -> factor 0.5 - cache_key = generate_recalibrate_orgs_cache_key(self.orgs[2].id) - val = redis_client.get(cache_key) - assert val is not None - assert float(val) == 0.5 - - @with_feature("organizations:dynamic-sampling") - @patch("sentry.quotas.backend.get_blended_sample_rate") - def test_recalibrate_orgs_multiple_orgs_with_different_volumes( - self, get_blended_sample_rate: MagicMock - ) -> None: - get_blended_sample_rate.return_value = 0.2 - org1 = self.create_old_organization("org-1") - org2 = self.create_old_organization("org-2") - project1 = self.create_old_project(name="project-1", organization=org1) - project2 = self.create_old_project(name="project-2", organization=org2) - - self.add_measure_metrics( - org1, - project1, - segment_keep=10, - segment_drop=90, - ) - self.add_measure_metrics( - org2, - project2, - segment_keep=50, - segment_drop=50, - ) - - self.set_sliding_window_org_sample_rate(org1.id, 0.2) - self.set_sliding_window_org_sample_rate(org2.id, 0.2) - - with self.tasks(): - recalibrate_orgs() - - redis_client = get_redis_client_for_ds() - org1_value = redis_client.get(generate_recalibrate_orgs_cache_key(org1.id)) - org2_value = redis_client.get(generate_recalibrate_orgs_cache_key(org2.id)) - - assert org1_value is not None - assert org2_value is not None - assert float(org1_value) == pytest.approx(2.0) - assert float(org2_value) == pytest.approx(0.4) - - -@freeze_time(MOCK_DATETIME) -class TestSlidingWindowOrgTask(TasksTestCase): - """Tests for the sliding_window_org task with measure parameter support.""" - - @property - def now(self): - return MOCK_DATETIME - - def setUp(self) -> None: - super().setUp() - self.orgs = [] - # Create orgs with different volumes - for i, volume in enumerate([100, 500, 1000]): - org = self.create_old_organization(f"test-org-{i}") - self.orgs.append(org) - project = self.create_old_project(name=f"test-project-{i}", organization=org) - - self.store_performance_metric( - name=SpanMRI.COUNT_PER_ROOT_PROJECT.value, - tags={"transaction": "foo", "is_segment": "true"}, - minutes_before_now=30, - value=volume, - project_id=project.id, - org_id=org.id, - ) - - @with_feature("organizations:dynamic-sampling") - @patch("sentry.dynamic_sampling.tasks.common.extrapolate_monthly_volume") - @patch("sentry.quotas.backend.get_transaction_sampling_tier_for_volume") - def test_sliding_window_org_processes_all_orgs_by_default( - self, - get_transaction_sampling_tier_for_volume: MagicMock, - extrapolate_monthly_volume: MagicMock, - ) -> None: - """ - Test that sliding_window_org processes all orgs using SEGMENTS measure by default. - """ - extrapolate_monthly_volume.side_effect = lambda volume, hours: volume - get_transaction_sampling_tier_for_volume.return_value = (1000, 0.25) - redis_client = get_redis_client_for_ds() - - with self.tasks(): - sliding_window_org() - - # All orgs should have cache entries - for org in self.orgs: - cache_key = generate_sliding_window_org_cache_key(org.id) - val = redis_client.get(cache_key) - assert val is not None, f"Org {org.id} should have sliding window cache entry" - - @with_feature("organizations:dynamic-sampling") - @override_options({"dynamic-sampling.legacy.killswitch": True}) - @patch("sentry.dynamic_sampling.tasks.common.extrapolate_monthly_volume") - @patch("sentry.quotas.backend.get_transaction_sampling_tier_for_volume") - def test_sliding_window_org_killswitch( - self, - get_transaction_sampling_tier_for_volume: MagicMock, - extrapolate_monthly_volume: MagicMock, - ) -> None: - extrapolate_monthly_volume.side_effect = lambda volume, hours: volume - get_transaction_sampling_tier_for_volume.return_value = (1000, 0.25) - redis_client = get_redis_client_for_ds() - - with self.tasks(): - sliding_window_org() - - for org in self.orgs: - assert redis_client.get(generate_sliding_window_org_cache_key(org.id)) is None From df6210639696ab8bb9d206feaa1914b6f6bd5938 Mon Sep 17 00:00:00 2001 From: Simon Hellmayr Date: Thu, 10 Sep 2026 14:12:35 +0200 Subject: [PATCH 3/3] test(dynamic-sampling): Keep the legacy pipeline tests with the per-org pipeline off Restore the legacy pipeline tests. They run the legacy pipeline end to end, so they switch the per-org pipeline off through its rollout options instead of relying on the former defaults. Co-Authored-By: Claude Fable 5.1 --- .../dynamic_sampling/tasks/test_tasks.py | 1067 +++++++++++++++++ 1 file changed, 1067 insertions(+) create mode 100644 tests/sentry/dynamic_sampling/tasks/test_tasks.py diff --git a/tests/sentry/dynamic_sampling/tasks/test_tasks.py b/tests/sentry/dynamic_sampling/tasks/test_tasks.py new file mode 100644 index 000000000000..b381f22951fc --- /dev/null +++ b/tests/sentry/dynamic_sampling/tasks/test_tasks.py @@ -0,0 +1,1067 @@ +from collections.abc import Callable +from datetime import timedelta +from unittest.mock import MagicMock, patch + +import pytest +from django.utils import timezone + +from sentry.dynamic_sampling import RuleType, generate_rules, get_redis_client_for_ds +from sentry.dynamic_sampling.per_org import cache as per_org_cache +from sentry.dynamic_sampling.rules.base import NEW_MODEL_THRESHOLD_IN_MINUTES +from sentry.dynamic_sampling.rules.biases.recalibration_bias import RecalibrationBias +from sentry.dynamic_sampling.tasks.boost_low_volume_projects import boost_low_volume_projects +from sentry.dynamic_sampling.tasks.boost_low_volume_transactions import ( + boost_low_volume_transactions, +) +from sentry.dynamic_sampling.tasks.helpers.boost_low_volume_projects import ( + generate_boost_low_volume_projects_cache_key, +) +from sentry.dynamic_sampling.tasks.helpers.boost_low_volume_transactions import ( + get_transactions_resampling_rates, +) +from sentry.dynamic_sampling.tasks.helpers.recalibrate_orgs import ( + generate_recalibrate_orgs_cache_key, + generate_recalibrate_projects_cache_key, +) +from sentry.dynamic_sampling.tasks.helpers.sliding_window import ( + generate_sliding_window_org_cache_key, + mark_sliding_window_org_executed, +) +from sentry.dynamic_sampling.tasks.recalibrate_orgs import recalibrate_orgs +from sentry.dynamic_sampling.tasks.sliding_window_org import sliding_window_org +from sentry.dynamic_sampling.types import DynamicSamplingMode +from sentry.snuba.metrics.naming_layer.mri import SpanMRI +from sentry.testutils.cases import BaseMetricsLayerTestCase, SnubaTestCase, TestCase +from sentry.testutils.helpers import with_feature +from sentry.testutils.helpers.datetime import freeze_time +from sentry.testutils.helpers.options import override_options + +MOCK_DATETIME = (timezone.now() - timedelta(days=1)).replace( + hour=0, minute=0, second=0, microsecond=0 +) + + +class TasksTestCase(BaseMetricsLayerTestCase, TestCase, SnubaTestCase): + def setUp(self) -> None: + super().setUp() + # These tests run the legacy pipeline end to end, so the per-org pipeline is + # switched off and rules read the legacy caches. + self.enterContext( + override_options( + { + "dynamic-sampling.per_org.rollout-rate": 0.0, + "dynamic-sampling.per_org.serving-rollout-rate": 0.0, + } + ) + ) + + @staticmethod + def old_date(): + return timezone.now() - timedelta(minutes=NEW_MODEL_THRESHOLD_IN_MINUTES + 1) + + @staticmethod + def disable_all_biases(project): + project.update_option( + "sentry:dynamic_sampling_biases", + [ + {"id": RuleType.BOOST_ENVIRONMENTS_RULE.value, "active": False}, + {"id": RuleType.IGNORE_HEALTH_CHECKS_RULE.value, "active": False}, + {"id": RuleType.BOOST_LATEST_RELEASES_RULE.value, "active": False}, + {"id": RuleType.BOOST_KEY_TRANSACTIONS_RULE.value, "active": False}, + {"id": RuleType.BOOST_LOW_VOLUME_TRANSACTIONS_RULE.value, "active": False}, + {"id": RuleType.BOOST_REPLAY_ID_RULE.value, "active": False}, + ], + ) + + def create_old_organization(self, name): + return self.create_organization(name=name, date_added=self.old_date()) + + def create_old_project(self, name, organization): + return self.create_project(name=name, organization=organization, date_added=self.old_date()) + + def create_project_and_add_metrics(self, name, count, org, tags=None, is_old=True): + if tags is None: + tags = {"transaction": "foo_transaction", "is_segment": "true"} + elif "is_segment" not in tags: + tags = {**tags, "is_segment": "true"} + + if is_old: + proj = self.create_old_project(name=name, organization=org) + else: + proj = self.create_project(name=name, organization=org) + + self.disable_all_biases(project=proj) + + self.store_performance_metric( + name=SpanMRI.COUNT_PER_ROOT_PROJECT.value, + tags=tags, + minutes_before_now=30, + value=count, + project_id=proj.id, + org_id=org.id, + ) + + return proj + + def create_project_without_metrics(self, name, org, is_old=True): + if is_old: + proj = self.create_old_project(name=name, organization=org) + else: + proj = self.create_project(name=name, organization=org) + + self.disable_all_biases(project=proj) + + return proj + + +@freeze_time(MOCK_DATETIME) +class TestBoostLowVolumeProjectsTasks(TasksTestCase): + @property + def now(self): + return MOCK_DATETIME + + @staticmethod + def add_sample_rate_per_project(org_id: int, project_id: int, sample_rate: float): + redis_client = get_redis_client_for_ds() + redis_client.hset( + name=generate_boost_low_volume_projects_cache_key(org_id), + key=str(project_id), + value=sample_rate, + ) + + @staticmethod + def sampling_tier_side_effect(*args, **kwargs): + volume = args[1] + + if volume == 20: + return 100_000, 0.25 + # We want to also hardcode the error case, to test how the system reacts to errors. + elif volume == 0: + return None + + return volume, 1.0 + + @staticmethod + def forecasted_volume_side_effect(*args, **kwargs): + return kwargs["volume"] + + @patch("sentry.quotas.backend.get_blended_sample_rate") + def test_boost_low_volume_projects_with_no_dynamic_sampling(self, get_blended_sample_rate): + get_blended_sample_rate.return_value = 0.25 + test_org = self.create_old_organization(name="sample-org") + + self.create_project_and_add_metrics("a", 9, test_org) + self.create_project_and_add_metrics("b", 7, test_org) + self.create_project_and_add_metrics("c", 3, test_org) + self.create_project_and_add_metrics("d", 1, test_org) + + with self.tasks(): + sliding_window_org() + boost_low_volume_projects() + + @with_feature("organizations:dynamic-sampling") + @patch("sentry.quotas.backend.get_blended_sample_rate") + def test_boost_low_volume_projects_simple( + self, + get_blended_sample_rate, + ): + get_blended_sample_rate.return_value = 0.25 + # Create a org + test_org = self.create_old_organization(name="sample-org") + + # Create 4 projects + proj_a = self.create_project_and_add_metrics("a", 9, test_org) + proj_b = self.create_project_and_add_metrics("b", 7, test_org) + proj_c = self.create_project_and_add_metrics("c", 3, test_org) + proj_d = self.create_project_and_add_metrics("d", 1, test_org) + + with self.tasks(): + sliding_window_org() + boost_low_volume_projects() + + # we expect only uniform rule + # also we test here that `generate_rules` can handle trough redis long floats + assert generate_rules(proj_a)[0]["samplingValue"] == { + "type": "sampleRate", + "value": pytest.approx(0.14814814814814817), + } + assert generate_rules(proj_b)[0]["samplingValue"] == { + "type": "sampleRate", + "value": pytest.approx(0.1904761904761905), + } + assert generate_rules(proj_c)[0]["samplingValue"] == { + "type": "sampleRate", + "value": pytest.approx(0.4444444444444444), + } + assert generate_rules(proj_d)[0]["samplingValue"] == {"type": "sampleRate", "value": 1.0} + + @with_feature("organizations:dynamic-sampling") + @override_options({"dynamic-sampling.legacy.killswitch": True}) + @patch("sentry.quotas.backend.get_blended_sample_rate") + def test_boost_low_volume_projects_killswitch(self, get_blended_sample_rate: MagicMock) -> None: + get_blended_sample_rate.return_value = 0.25 + test_org = self.create_old_organization(name="sample-org") + self.create_project_and_add_metrics("a", 9, test_org) + self.create_project_and_add_metrics("b", 1, test_org) + + with self.tasks(): + boost_low_volume_projects() + + redis_client = get_redis_client_for_ds() + assert redis_client.hgetall(generate_boost_low_volume_projects_cache_key(test_org.id)) == {} + + @with_feature("organizations:dynamic-sampling") + @patch("sentry.quotas.backend.get_blended_sample_rate") + def test_boost_low_volume_projects_simple_with_empty_project( + self, + get_blended_sample_rate, + ): + get_blended_sample_rate.return_value = 0.25 + test_org = self.create_old_organization(name="sample-org") + + proj_a = self.create_project_and_add_metrics("a", 9, test_org) + proj_b = self.create_project_and_add_metrics("b", 7, test_org) + proj_c = self.create_project_and_add_metrics("c", 3, test_org) + proj_d = self.create_project_and_add_metrics("d", 1, test_org) + proj_e = self.create_project_without_metrics("e", test_org) + + with self.tasks(): + sliding_window_org() + boost_low_volume_projects() + + # we expect only uniform rule + # also we test here that `generate_rules` can handle trough redis long floats + assert generate_rules(proj_a)[0]["samplingValue"] == { + "type": "sampleRate", + "value": pytest.approx(0.14814814814814817), + } + assert generate_rules(proj_b)[0]["samplingValue"] == { + "type": "sampleRate", + "value": pytest.approx(0.1904761904761905), + } + assert generate_rules(proj_c)[0]["samplingValue"] == { + "type": "sampleRate", + "value": pytest.approx(0.4444444444444444), + } + assert generate_rules(proj_d)[0]["samplingValue"] == {"type": "sampleRate", "value": 1.0} + assert generate_rules(proj_e)[0]["samplingValue"] == {"type": "sampleRate", "value": 1.0} + + @with_feature("organizations:dynamic-sampling") + @patch("sentry.quotas.backend.get_blended_sample_rate") + @patch("sentry.quotas.backend.get_transaction_sampling_tier_for_volume") + @patch("sentry.dynamic_sampling.tasks.common.extrapolate_monthly_volume") + def test_boost_low_volume_projects_simple_with_sliding_window_org_from_cache( + self, + extrapolate_monthly_volume, + get_transaction_sampling_tier_for_volume, + get_blended_sample_rate, + ): + extrapolate_monthly_volume.side_effect = self.forecasted_volume_side_effect + get_transaction_sampling_tier_for_volume.side_effect = self.sampling_tier_side_effect + get_blended_sample_rate.return_value = 0.8 + + test_org = self.create_old_organization(name="sample-org") + + proj_a = self.create_project_and_add_metrics("a", 9, test_org) + proj_b = self.create_project_and_add_metrics("b", 7, test_org) + proj_c = self.create_project_and_add_metrics("c", 3, test_org) + proj_d = self.create_project_and_add_metrics("d", 1, test_org) + + with self.tasks(): + sliding_window_org() + boost_low_volume_projects() + + # we expect only uniform rule + # also we test here that `generate_rules` can handle trough redis long floats + assert generate_rules(proj_a)[0]["samplingValue"] == { + "type": "sampleRate", + "value": pytest.approx(0.14814814814814817), + } + assert generate_rules(proj_b)[0]["samplingValue"] == { + "type": "sampleRate", + "value": pytest.approx(0.1904761904761905), + } + assert generate_rules(proj_c)[0]["samplingValue"] == { + "type": "sampleRate", + "value": pytest.approx(0.4444444444444444), + } + assert generate_rules(proj_d)[0]["samplingValue"] == {"type": "sampleRate", "value": 1.0} + + @with_feature("organizations:dynamic-sampling") + @patch( + "sentry.dynamic_sampling.tasks.boost_low_volume_projects.schedule_invalidate_project_config" + ) + @patch("sentry.quotas.backend.get_blended_sample_rate") + @patch("sentry.quotas.backend.get_transaction_sampling_tier_for_volume") + @patch("sentry.dynamic_sampling.tasks.common.extrapolate_monthly_volume") + def test_config_invalidation_when_sample_rates_change( + self, + extrapolate_monthly_volume, + get_transaction_sampling_tier_for_volume, + get_blended_sample_rate, + schedule_invalidate_project_config, + ): + extrapolate_monthly_volume.side_effect = self.forecasted_volume_side_effect + get_transaction_sampling_tier_for_volume.side_effect = self.sampling_tier_side_effect + get_blended_sample_rate.return_value = 0.8 + + test_org = self.create_old_organization(name="sample-org") + + proj_a = self.create_project_and_add_metrics("a", 9, test_org) + proj_b = self.create_project_and_add_metrics("b", 7, test_org) + + self.add_sample_rate_per_project(org_id=test_org.id, project_id=proj_a.id, sample_rate=0.1) + self.add_sample_rate_per_project(org_id=test_org.id, project_id=proj_b.id, sample_rate=0.2) + + with self.tasks(): + sliding_window_org() + boost_low_volume_projects() + + assert schedule_invalidate_project_config.call_count == 2 + + @with_feature("organizations:dynamic-sampling") + @patch( + "sentry.dynamic_sampling.tasks.boost_low_volume_projects.schedule_invalidate_project_config" + ) + @patch("sentry.quotas.backend.get_blended_sample_rate") + @patch("sentry.quotas.backend.get_transaction_sampling_tier_for_volume") + @patch("sentry.dynamic_sampling.tasks.common.extrapolate_monthly_volume") + def test_config_invalidation_when_sample_rates_do_not_change( + self, + extrapolate_monthly_volume, + get_transaction_sampling_tier_for_volume, + get_blended_sample_rate, + schedule_invalidate_project_config, + ): + extrapolate_monthly_volume.side_effect = self.forecasted_volume_side_effect + get_transaction_sampling_tier_for_volume.side_effect = self.sampling_tier_side_effect + get_blended_sample_rate.return_value = 1.0 + + test_org = self.create_old_organization(name="sample-org") + + proj_a = self.create_project_and_add_metrics("a", 9, test_org) + proj_b = self.create_project_and_add_metrics("b", 7, test_org) + + self.add_sample_rate_per_project(org_id=test_org.id, project_id=proj_a.id, sample_rate=1.0) + self.add_sample_rate_per_project(org_id=test_org.id, project_id=proj_b.id, sample_rate=1.0) + + with self.tasks(): + boost_low_volume_projects() + + schedule_invalidate_project_config.assert_not_called() + + +@freeze_time(MOCK_DATETIME) +class TestBoostLowVolumeTransactionsTasks(TasksTestCase): + @property + def now(self): + return MOCK_DATETIME + + def setUp(self) -> None: + super().setUp() + self.orgs_info = [] + num_orgs = 3 + num_proj_per_org = 3 + for org_idx in range(num_orgs): + org = self.create_old_organization(f"test-org{org_idx}") + org_info = {"org_id": org.id, "project_ids": []} + self.orgs_info.append(org_info) + for proj_idx in range(num_proj_per_org): + p = self.create_old_project(name=f"test-project-{proj_idx}", organization=org) + org_info["project_ids"].append(p.id) + # create 5 transaction types + for name in ["ts1", "ts2", "tm3", "tl4", "tl5"]: + # make up some unique count + idx = org_idx * num_orgs + proj_idx + num_transactions = self.get_count_for_transaction(idx, name) + self.store_performance_metric( + name=SpanMRI.COUNT_PER_ROOT_PROJECT.value, + tags={"transaction": name, "is_segment": "true"}, + minutes_before_now=30, + value=num_transactions, + project_id=p.id, + org_id=org.id, + ) + self.org_ids = [org["org_id"] for org in self.orgs_info] + + def get_count_for_transaction(self, idx: int, name: str): + """ + Create some known count based on transaction name and the order (based on org and project) + """ + counts = { + "ts1": 1, + "ts2": 100, + "tm3": 1000, + "tl4": 2000, + "tl5": 3000, + } + return idx + counts[name] + + @staticmethod + def flush_redis(): + get_redis_client_for_ds().flushdb() + + @staticmethod + def set_boost_low_volume_projects_cache_entry(org_id: int, project_id: int, value: str): + redis = get_redis_client_for_ds() + cache_key = generate_boost_low_volume_projects_cache_key(org_id=org_id) + redis.hset(name=cache_key, key=str(project_id), value=value) + + def set_boost_low_volume_projects_sample_rate( + self, org_id: int, project_id: int, sample_rate: float + ): + self.set_boost_low_volume_projects_cache_entry(org_id, project_id, str(sample_rate)) + + def set_prioritise_by_project_invalid(self, org_id: int, project_id: int): + # We want also to test for this case in order to verify the fallback to the `get_blended_sample_rate`. + self.set_boost_low_volume_projects_cache_entry(org_id, project_id, "invalid") + + def for_all_orgs_and_projects(self, block: Callable[[int, int], None]): + for org in self.orgs_info: + org_id = org["org_id"] + for project_id in org["project_ids"]: + block(org_id, project_id) + + def set_boost_low_volume_projects_invalid_for_all(self): + self.for_all_orgs_and_projects( + lambda org_id, project_id: self.set_prioritise_by_project_invalid(org_id, project_id) + ) + + def set_boost_low_volume_projects_for_all(self, sample_rate: float): + self.for_all_orgs_and_projects( + lambda org_id, project_id: self.set_boost_low_volume_projects_sample_rate( + org_id, project_id, sample_rate + ) + ) + + @patch("sentry.quotas.backend.get_blended_sample_rate") + def test_boost_low_volume_transactions_with_blended_sample_rate_and_no_dynamic_sampling( + self, get_blended_sample_rate + ): + """ + Create orgs projects & transactions and then check that the rebalancing model is not called because dynamic + sampling is disabled + """ + BLENDED_RATE = 0.25 + get_blended_sample_rate.return_value = BLENDED_RATE + + with self.tasks(): + boost_low_volume_transactions() + + @with_feature("organizations:dynamic-sampling") + @patch("sentry.quotas.backend.get_blended_sample_rate") + def test_boost_low_volume_transactions_with_blended_sample_rate( + self, get_blended_sample_rate: MagicMock + ) -> None: + """ + Create orgs projects & transactions and then check that the task creates rebalancing data + in Redis. + """ + BLENDED_RATE = 0.25 + get_blended_sample_rate.return_value = BLENDED_RATE + + with self.tasks(): + boost_low_volume_transactions() + + # now redis should contain rebalancing data for our projects + for org in self.orgs_info: + org_id = org["org_id"] + for proj_id in org["project_ids"]: + tran_rate, global_rate = get_transactions_resampling_rates( + org_id=org_id, proj_id=proj_id, default_rate=0.1 + ) + for transaction_name in ["ts1", "ts2", "tm3", "tl4", "tl5"]: + assert ( + transaction_name in tran_rate + ) # check we have some rate calculated for each transaction + assert global_rate == BLENDED_RATE + + @with_feature("organizations:dynamic-sampling") + @override_options({"dynamic-sampling.legacy.killswitch": True}) + @patch("sentry.quotas.backend.get_blended_sample_rate") + def test_boost_low_volume_transactions_killswitch( + self, get_blended_sample_rate: MagicMock + ) -> None: + get_blended_sample_rate.return_value = 0.25 + + with self.tasks(): + boost_low_volume_transactions() + + for org in self.orgs_info: + for proj_id in org["project_ids"]: + tran_rate, _ = get_transactions_resampling_rates( + org_id=org["org_id"], proj_id=proj_id, default_rate=0.1 + ) + assert tran_rate == {} + + @with_feature("organizations:dynamic-sampling") + @patch("sentry.quotas.backend.get_blended_sample_rate") + def test_boost_low_volume_transactions_with_sliding_window_org( + self, get_blended_sample_rate: MagicMock + ) -> None: + """ + Create orgs projects & transactions and then check that the task creates rebalancing data + in Redis with the sliding window per org enabled. + """ + BLENDED_RATE = 0.25 + get_blended_sample_rate.return_value = BLENDED_RATE + + for sliding_window_step, used_sample_rate in ((1, 1.0), (2, BLENDED_RATE), (3, 0.5)): + # We flush redis after each run, to make sure no data persists. + self.flush_redis() + + # No value in cache and sliding window org executed. + if sliding_window_step == 1: + mark_sliding_window_org_executed() + # Invalid value in cache. + elif sliding_window_step == 2: + self.set_boost_low_volume_projects_invalid_for_all() + # Value in cache. + elif sliding_window_step == 3: + self.set_boost_low_volume_projects_for_all(used_sample_rate) + + with self.tasks(): + boost_low_volume_transactions() + + # now redis should contain rebalancing data for our projects + for org in self.orgs_info: + org_id = org["org_id"] + for proj_id in org["project_ids"]: + tran_rate, global_rate = get_transactions_resampling_rates( + org_id=org_id, proj_id=proj_id, default_rate=0.1 + ) + + if sliding_window_step == 1: + # If the sample rate is 100%, we will not find anything in cache, since we don't + # need to run and store the rebalancing. + assert tran_rate == {} + else: + # If the sample rate is < 100%, we want to check that in cache we have a value with + # the correct global rate. + for transaction_name in ["ts1", "ts2", "tm3", "tl4", "tl5"]: + assert ( + transaction_name in tran_rate + ) # check we have some rate calculated for each transaction + + assert global_rate == used_sample_rate + + @with_feature("organizations:dynamic-sampling") + @patch("sentry.quotas.backend.get_blended_sample_rate") + def test_boost_low_volume_transactions_partial( + self, get_blended_sample_rate: MagicMock + ) -> None: + """ + Test the V2 algorithm is used, only specified projects are balanced and the + rest get a global rate + + Create orgs projects & transactions and then check that the task creates rebalancing data + in Redis + """ + BLENDED_RATE = 0.25 + get_blended_sample_rate.return_value = BLENDED_RATE + + with self.options( + { + "dynamic-sampling.prioritise_transactions.num_explicit_large_transactions": 1, + "dynamic-sampling.prioritise_transactions.rebalance_intensity": 0.7, + } + ): + with self.tasks(): + boost_low_volume_transactions() + + # now redis should contain rebalancing data for our projects + for org in self.orgs_info: + org_id = org["org_id"] + for proj_id in org["project_ids"]: + tran_rate, implicit_rate = get_transactions_resampling_rates( + org_id=org_id, proj_id=proj_id, default_rate=0.1 + ) + # explicit transactions + for transaction_name in ["tl5"]: + assert ( + transaction_name in tran_rate + ) # check we have some rate calculated for each transaction + # implicit transactions + for transaction_name in ["ts1", "ts2", "tm3", "tl4"]: + assert ( + transaction_name not in tran_rate + ) # check we have some rate calculated for each transaction + # we do have some different rate for implicit transactions + assert implicit_rate != BLENDED_RATE + + +@freeze_time(MOCK_DATETIME) +class TestRecalibrateOrgsTasks(TasksTestCase): + @property + def now(self): + return MOCK_DATETIME + + def setUp(self) -> None: + super().setUp() + self.orgs_info = [] + self.orgs = [] + self.num_proj = 2 + self.orgs_sampling = [10, 20, 40] + # create some orgs, projects and transactions + for org_rate in self.orgs_sampling: + org = self.create_old_organization(f"test-org-{org_rate}") + org_info = {"org_id": org.id, "project_ids": [], "projects": []} + self.orgs_info.append(org_info) + self.orgs.append(org) + for proj_idx in range(self.num_proj): + p = self.create_old_project(name=f"test-project-{proj_idx}", organization=org) + org_info["projects"].append(p) + org_info["project_ids"].append(p.id) + self.add_metrics(org, p, org_rate) + + def add_metrics(self, org, project, sample_rate): + base_tags = {"transaction": "trans-x", "is_segment": "true"} + + if sample_rate < 100: + self.store_performance_metric( + name=SpanMRI.COUNT_PER_ROOT_PROJECT.value, + tags={**base_tags, "decision": "drop"}, + minutes_before_now=2, + value=100 - sample_rate, + project_id=project.id, + org_id=org.id, + ) + if sample_rate > 0: + self.store_performance_metric( + name=SpanMRI.COUNT_PER_ROOT_PROJECT.value, + tags={**base_tags, "decision": "keep"}, + minutes_before_now=2, + value=sample_rate, + project_id=project.id, + org_id=org.id, + ) + + def add_measure_metrics( + self, + org, + project, + *, + segment_keep: int, + segment_drop: int, + ) -> None: + segment_tags = {"transaction": "trans-x", "is_segment": "true"} + if segment_drop: + self.store_performance_metric( + name=SpanMRI.COUNT_PER_ROOT_PROJECT.value, + tags={**segment_tags, "decision": "drop"}, + minutes_before_now=2, + value=segment_drop, + project_id=project.id, + org_id=org.id, + ) + if segment_keep: + self.store_performance_metric( + name=SpanMRI.COUNT_PER_ROOT_PROJECT.value, + tags={**segment_tags, "decision": "keep"}, + minutes_before_now=2, + value=segment_keep, + project_id=project.id, + org_id=org.id, + ) + + @staticmethod + def set_sliding_window_org_cache_entry(org_id: int, value: str): + redis = get_redis_client_for_ds() + cache_key = generate_sliding_window_org_cache_key(org_id=org_id) + redis.set(cache_key, value) + + def set_sliding_window_org_sample_rate(self, org_id: int, sample_rate: float): + self.set_sliding_window_org_cache_entry(org_id, str(sample_rate)) + + def for_all_orgs(self, block: Callable[[int], None]): + for org in self.orgs_info: + org_id = org["org_id"] + block(org_id) + + def set_sliding_window_org_sample_rate_for_all(self, sample_rate: float): + self.for_all_orgs( + lambda org_id: self.set_sliding_window_org_sample_rate(org_id, sample_rate) + ) + + @patch("sentry.dynamic_sampling.tasks.recalibrate_orgs.compute_adjusted_factor") + @patch("sentry.quotas.backend.get_blended_sample_rate") + def test_recalibrate_orgs_with_no_dynamic_sampling( + self, get_blended_sample_rate, computed_adjusted_factor + ): + """ + Test that the recalibration of orgs doesn't happen if dynamic sampling is not enabled + """ + get_blended_sample_rate.return_value = 0.1 + self.set_sliding_window_org_sample_rate_for_all(0.2) + + with self.tasks(): + recalibrate_orgs() + + computed_adjusted_factor.assert_not_called() + + @with_feature("organizations:dynamic-sampling") + @patch("sentry.quotas.backend.get_blended_sample_rate") + def test_recalibrate_orgs_with_sliding_window_org( + self, get_blended_sample_rate: MagicMock + ) -> None: + """ + Test that the org are going to be rebalanced at 20% and that the sample rate used is the one from the sliding + window org. + + The first org is 10%, so we should increase the sampling + The second org is at 20%, so we are spot on + The third is at 40%, so we should decrease the sampling + """ + get_blended_sample_rate.return_value = 0.1 + self.set_sliding_window_org_sample_rate_for_all(0.2) + + redis_client = get_redis_client_for_ds() + + with self.tasks(): + recalibrate_orgs() + + for idx, org in enumerate(self.orgs): + cache_key = generate_recalibrate_orgs_cache_key(org.id) + val = redis_client.get(cache_key) + + if idx == 0: + assert val is not None + # we sampled at 10% half of what we want so we should adjust by 2 + assert float(val) == 2.0 + elif idx == 1: + # we sampled at 20% we should be spot on (no adjustment) + assert val is None + elif idx == 2: + assert val is not None + # we sampled at 40% twice as much as we wanted we should adjust by 0.5 + assert float(val) == 0.5 + + # now if we run it again (with the same data in the database, the algorithm + # should double down... the previous factor didn't do anything so apply it again) + with self.tasks(): + recalibrate_orgs() + + for idx, org in enumerate(self.orgs): + cache_key = generate_recalibrate_orgs_cache_key(org.id) + val = redis_client.get(cache_key) + + if idx == 0: + assert val is not None + # we sampled at 10% when already having a factor of two half of what we want so we + # should double the current factor to 4 + assert float(val) == 4.0 + elif idx == 1: + # we sampled at 20% we should be spot on (no adjustment) + assert val is None + elif idx == 2: + assert val is not None + # we sampled at 40% twice as much as we wanted we already have a factor of 0.5 + # half it again to 0.25 + assert float(val) == 0.25 + + @with_feature("organizations:dynamic-sampling") + @override_options({"dynamic-sampling.legacy.killswitch": True}) + @patch("sentry.quotas.backend.get_blended_sample_rate") + def test_recalibrate_orgs_killswitch(self, get_blended_sample_rate: MagicMock) -> None: + get_blended_sample_rate.return_value = 0.1 + self.set_sliding_window_org_sample_rate_for_all(0.2) + + with self.tasks(): + recalibrate_orgs() + + redis_client = get_redis_client_for_ds() + for org in self.orgs: + assert redis_client.get(generate_recalibrate_orgs_cache_key(org.id)) is None + + @with_feature("organizations:dynamic-sampling") + @patch("sentry.quotas.backend.get_blended_sample_rate") + def test_recalibrate_orgs_skips_orgs_served_the_per_org_factor( + self, get_blended_sample_rate: MagicMock + ) -> None: + """An org served the per-org factor keeps the legacy cache untouched. + + Both loops step by previous * target / measured. Writing here as well would step a + factor nothing applies, so it would walk to a rebalance bound. + """ + get_blended_sample_rate.return_value = 0.1 + self.set_sliding_window_org_sample_rate_for_all(0.2) + + served_org = self.orgs[0] + redis_client = get_redis_client_for_ds() + + with ( + override_options({"dynamic-sampling.per_org.serving-org-ids": [served_org.id]}), + self.tasks(), + ): + recalibrate_orgs() + + # The served org sampled at 10% against a 20% target, so the legacy task would have + # written a factor of 2.0 here. + assert redis_client.get(generate_recalibrate_orgs_cache_key(served_org.id)) is None + + # Every other org still has its factor written by the legacy task. + other_factor = redis_client.get(generate_recalibrate_orgs_cache_key(self.orgs[2].id)) + assert other_factor is not None + assert float(other_factor) == 0.5 + + @with_feature("organizations:dynamic-sampling") + def test_recalibrate_orgs_continues_from_the_per_org_factor_after_switching_back( + self, + ) -> None: + """An org switched back to the legacy pipeline steps from the factor it was served. + + The legacy key expired while the per-org pipeline served the org, so without the + carry-over the correction would restart from 1.0 and drop the whole boost. + """ + self.set_sliding_window_org_sample_rate_for_all(0.2) + + # This org stored metrics at a 10% sampling rate, so it measures at 0.1. + switched_back_org = self.orgs[0] + per_org_cache.set_adjusted_factor(switched_back_org.id, 3.0) + + redis_client = get_redis_client_for_ds() + + with self.tasks(): + recalibrate_orgs() + + # 3.0 * (0.2 target / 0.1 measured), instead of the 2.0 a restart from 1.0 gives. + factor = redis_client.get(generate_recalibrate_orgs_cache_key(switched_back_org.id)) + assert factor is not None + assert float(factor) == 6.0 + + @with_feature("organizations:dynamic-sampling") + @with_feature("organizations:dynamic-sampling-custom") + def test_recalibrate_orgs_with_custom_ds(self) -> None: + """ + Test several organizations with mixed sampling mode. + + The first org is 10%, so we should increase the sampling + The second org is at 20%, so we are spot on + The third is at 40%, so we should decrease the sampling + """ + + # First two orgs have a 20% sample rate configured, third one is in project mode + self.orgs[0].update_option("sentry:target_sample_rate", 0.2) + self.orgs[1].update_option("sentry:target_sample_rate", 0.2) + self.orgs[2].update_option("sentry:sampling_mode", DynamicSamplingMode.PROJECT) + + # First project gets same 20% sample rate, the other one stays at implicit 100% + p1, p2 = self.orgs_info[2]["projects"] + p1.update_option("sentry:target_sample_rate", 0.2) + + with self.tasks(): + recalibrate_orgs() + + redis_client = get_redis_client_for_ds() + + # First org was sampled at 10%, should be recalibrated at 2x to 20%. + assert redis_client.get(generate_recalibrate_orgs_cache_key(self.orgs[0].id)) == "2.0" + # Second org was sampled at 20%, should not be recalibrated. + assert redis_client.get(generate_recalibrate_orgs_cache_key(self.orgs[1].id)) is None + + # Third org should not have org-level recalibration. + assert redis_client.get(generate_recalibrate_orgs_cache_key(self.orgs[2].id)) is None + # First project was sampled at 40%, should be recalibrated at 0.5x to 20%. + assert redis_client.get(generate_recalibrate_projects_cache_key(p1.id)) == "0.5" + # Second project was sampled at 40%, should be recalibrated at 2.5x to 100%. + assert redis_client.get(generate_recalibrate_projects_cache_key(p2.id)) == "2.5" + + assert RecalibrationBias().generate_rules(p1, base_sample_rate=1.0) == [ + { + "samplingValue": {"type": "factor", "value": 0.5}, + "type": "trace", + "condition": {"op": "and", "inner": []}, + "id": 1004, + } + ] + + assert RecalibrationBias().generate_rules(p2, base_sample_rate=1.0) == [ + { + "samplingValue": {"type": "factor", "value": 2.5}, + "type": "trace", + "condition": {"op": "and", "inner": []}, + "id": 1004, + } + ] + + @with_feature("organizations:dynamic-sampling") + @patch("sentry.quotas.backend.get_blended_sample_rate") + def test_rules_generation_with_recalibrate_orgs( + self, get_blended_sample_rate: MagicMock + ) -> None: + """ + Test that we pass rebalancing values all the way to the rules. + """ + get_blended_sample_rate.return_value = 0.20 + + with self.tasks(): + recalibrate_orgs() + + for org_idx, org in enumerate(self.orgs): + for project in org.project_set.all(): + rules = RecalibrationBias().generate_rules(project, base_sample_rate=0.5) + if org_idx == 0: + # we sampled at 10% half of what we want so we should adjust by 2 + assert rules == [ + { + "samplingValue": {"type": "factor", "value": 2.0}, + "type": "trace", + "condition": {"op": "and", "inner": []}, + "id": 1004, + } + ] + elif org_idx == 1: + # we sampled at 20% we should be spot on (no rule) + assert rules == [] + elif org_idx == 2: + # we sampled at 40% twice as much as we wanted we should adjust by 0.5 + assert rules == [ + { + "samplingValue": {"type": "factor", "value": 0.5}, + "type": "trace", + "condition": {"op": "and", "inner": []}, + "id": 1004, + } + ] + + @with_feature("organizations:dynamic-sampling") + @patch("sentry.quotas.backend.get_blended_sample_rate") + def test_recalibrate_orgs_uses_segments_measure( + self, get_blended_sample_rate: MagicMock + ) -> None: + """ + Test that all orgs use segment metrics (SEGMENTS measure) for recalibration. + """ + get_blended_sample_rate.return_value = 0.1 + self.set_sliding_window_org_sample_rate_for_all(0.2) + + redis_client = get_redis_client_for_ds() + + with self.tasks(): + recalibrate_orgs() + + # First org should be recalibrated (sampled at 10%, target 20% -> factor 2.0) + cache_key = generate_recalibrate_orgs_cache_key(self.orgs[0].id) + val = redis_client.get(cache_key) + assert val is not None + assert float(val) == 2.0 + + # Second org sampled at 20%, target 20% -> no adjustment needed + cache_key = generate_recalibrate_orgs_cache_key(self.orgs[1].id) + val = redis_client.get(cache_key) + assert val is None + + # Third org sampled at 40%, target 20% -> factor 0.5 + cache_key = generate_recalibrate_orgs_cache_key(self.orgs[2].id) + val = redis_client.get(cache_key) + assert val is not None + assert float(val) == 0.5 + + @with_feature("organizations:dynamic-sampling") + @patch("sentry.quotas.backend.get_blended_sample_rate") + def test_recalibrate_orgs_multiple_orgs_with_different_volumes( + self, get_blended_sample_rate: MagicMock + ) -> None: + get_blended_sample_rate.return_value = 0.2 + org1 = self.create_old_organization("org-1") + org2 = self.create_old_organization("org-2") + project1 = self.create_old_project(name="project-1", organization=org1) + project2 = self.create_old_project(name="project-2", organization=org2) + + self.add_measure_metrics( + org1, + project1, + segment_keep=10, + segment_drop=90, + ) + self.add_measure_metrics( + org2, + project2, + segment_keep=50, + segment_drop=50, + ) + + self.set_sliding_window_org_sample_rate(org1.id, 0.2) + self.set_sliding_window_org_sample_rate(org2.id, 0.2) + + with self.tasks(): + recalibrate_orgs() + + redis_client = get_redis_client_for_ds() + org1_value = redis_client.get(generate_recalibrate_orgs_cache_key(org1.id)) + org2_value = redis_client.get(generate_recalibrate_orgs_cache_key(org2.id)) + + assert org1_value is not None + assert org2_value is not None + assert float(org1_value) == pytest.approx(2.0) + assert float(org2_value) == pytest.approx(0.4) + + +@freeze_time(MOCK_DATETIME) +class TestSlidingWindowOrgTask(TasksTestCase): + """Tests for the sliding_window_org task with measure parameter support.""" + + @property + def now(self): + return MOCK_DATETIME + + def setUp(self) -> None: + super().setUp() + self.orgs = [] + # Create orgs with different volumes + for i, volume in enumerate([100, 500, 1000]): + org = self.create_old_organization(f"test-org-{i}") + self.orgs.append(org) + project = self.create_old_project(name=f"test-project-{i}", organization=org) + + self.store_performance_metric( + name=SpanMRI.COUNT_PER_ROOT_PROJECT.value, + tags={"transaction": "foo", "is_segment": "true"}, + minutes_before_now=30, + value=volume, + project_id=project.id, + org_id=org.id, + ) + + @with_feature("organizations:dynamic-sampling") + @patch("sentry.dynamic_sampling.tasks.common.extrapolate_monthly_volume") + @patch("sentry.quotas.backend.get_transaction_sampling_tier_for_volume") + def test_sliding_window_org_processes_all_orgs_by_default( + self, + get_transaction_sampling_tier_for_volume: MagicMock, + extrapolate_monthly_volume: MagicMock, + ) -> None: + """ + Test that sliding_window_org processes all orgs using SEGMENTS measure by default. + """ + extrapolate_monthly_volume.side_effect = lambda volume, hours: volume + get_transaction_sampling_tier_for_volume.return_value = (1000, 0.25) + redis_client = get_redis_client_for_ds() + + with self.tasks(): + sliding_window_org() + + # All orgs should have cache entries + for org in self.orgs: + cache_key = generate_sliding_window_org_cache_key(org.id) + val = redis_client.get(cache_key) + assert val is not None, f"Org {org.id} should have sliding window cache entry" + + @with_feature("organizations:dynamic-sampling") + @override_options({"dynamic-sampling.legacy.killswitch": True}) + @patch("sentry.dynamic_sampling.tasks.common.extrapolate_monthly_volume") + @patch("sentry.quotas.backend.get_transaction_sampling_tier_for_volume") + def test_sliding_window_org_killswitch( + self, + get_transaction_sampling_tier_for_volume: MagicMock, + extrapolate_monthly_volume: MagicMock, + ) -> None: + extrapolate_monthly_volume.side_effect = lambda volume, hours: volume + get_transaction_sampling_tier_for_volume.return_value = (1000, 0.25) + redis_client = get_redis_client_for_ds() + + with self.tasks(): + sliding_window_org() + + for org in self.orgs: + assert redis_client.get(generate_sliding_window_org_cache_key(org.id)) is None