diff --git a/sentry-options/schemas/snuba/schema.json b/sentry-options/schemas/snuba/schema.json index c6bbb6cbf2..1005037938 100644 --- a/sentry-options/schemas/snuba/schema.json +++ b/sentry-options/schemas/snuba/schema.json @@ -386,21 +386,6 @@ "max_threads": { "type": "integer", "optional": true - }, - "cross_org_referrer_limits": { - "type": "object", - "additionalProperties": { - "type": "object", - "properties": { - "max_threads": { - "type": "integer" - }, - "concurrent_limit": { - "type": "integer" - } - } - }, - "optional": true } }, "additionalProperties": { diff --git a/snuba/datasets/configuration/outcomes/entities/outcomes_cross_org_hourly.yaml b/snuba/datasets/configuration/outcomes/entities/outcomes_cross_org_hourly.yaml index 49f6134951..d3800446bd 100644 --- a/snuba/datasets/configuration/outcomes/entities/outcomes_cross_org_hourly.yaml +++ b/snuba/datasets/configuration/outcomes/entities/outcomes_cross_org_hourly.yaml @@ -31,7 +31,6 @@ query_processors: # Dedicated org-level outcomes entity for cross-org active-org discovery. Unlike the # shared `outcomes` entity it intentionally has an EMPTY validators list (no # EntityRequiredColumnValidator for org_id), so callers can group by org_id without a -# per-org filter. Access is governed at the storage layer by -# CrossOrgQueryAllocationPolicy. +# per-org filter. validators: [] validate_data_model: error diff --git a/snuba/datasets/configuration/outcomes/storages/cross_org_hourly.yaml b/snuba/datasets/configuration/outcomes/storages/cross_org_hourly.yaml index f29fcae042..44ed2bc2b0 100644 --- a/snuba/datasets/configuration/outcomes/storages/cross_org_hourly.yaml +++ b/snuba/datasets/configuration/outcomes/storages/cross_org_hourly.yaml @@ -35,5 +35,4 @@ query_processors: # shared outcomes_hourly storage over the same ClickHouse tables but intentionally # OMITS the OrgIdEnforcer mandatory_condition_checker so a query can scan across # organizations. The shared `outcomes` storage keeps OrgIdEnforcer; this storage is -# reachable only via the dedicated `outcomes_cross_org_hourly` entity and is gated by -# CrossOrgQueryAllocationPolicy (registered referrers only). +# reachable only via the dedicated `outcomes_cross_org_hourly` entity. diff --git a/snuba/query/allocation_policies/cross_org.py b/snuba/query/allocation_policies/cross_org.py deleted file mode 100644 index 2f582960c1..0000000000 --- a/snuba/query/allocation_policies/cross_org.py +++ /dev/null @@ -1,176 +0,0 @@ -from __future__ import annotations - -import logging -import typing -from typing import Any, cast - -from snuba.configs.configuration import Configuration, ResourceIdentifier -from snuba.query.allocation_policies import ( - MAX_THRESHOLD, - NO_SUGGESTION, - NO_UNITS, - QueryResultOrError, - QuotaAllowance, -) -from snuba.query.allocation_policies.concurrent_rate_limit import ( - BaseConcurrentRateLimitAllocationPolicy, -) -from snuba.redis import RedisClientKey, get_redis_client -from snuba.state.rate_limit import RateLimitParameters -from snuba.utils.serializable_exception import JsonSerializable - -DEFAULT_CONCURRENT_QUERIES_LIMIT = 22 -DEFAULT_PER_SECOND_QUERIES_LIMIT = 50 - -rds = get_redis_client(RedisClientKey.RATE_LIMITER) - -logger = logging.getLogger("snuba.query.allocation_policy_cross_org") - -_RATE_LIMIT_NAME = "concurrent_limit_policy" -_UNREGISTERED_REFERRER_MAX_THREADS = 1 -_UNREGISTERED_REFERRER_CONCURRENT_QUERIES = 1 - -QUOTA_UNIT = "concurrent_queries" -SUGGESTION = "scan less concurrent queries" - - -class CrossOrgQueryAllocationPolicy(BaseConcurrentRateLimitAllocationPolicy): - """A case-by-case allocation policy for cross-org queries. All referrers affected by this policy have to be registered - in this class's configuration through the `cross_org_referrer_limits` parameter. Example: - - ```yaml - - name: CrossOrgQueryAllocationPolicy - is_enforced: 0 - cross_org_referrer_limits: - dynamic_sampling.counters.get_org_transaction_volumes: - max_threads: 4 - concurrent_limit: 10 - ``` - - Each referrer gets a concurrent limit (applied per referrer) and a max_threads limit (applied to every query made by that referrer). - Both limits are static but changeable at runtime. - - unregistered referrers are assigned a default concurrent limit and max_threads limit of 1. - """ - - @property - def rate_limit_name(self) -> str: - return "cross_org_query_policy" - - def _additional_config_definitions(self) -> list[Configuration]: - return super()._additional_config_definitions() + [ - Configuration( - name="referrer_concurrent_override", - description="""override the concurrent limit for a referrer""", - value_type=int, - param_types={"referrer": str}, - default=-1, - ), - ] - - def _validate_cross_org_referrer_limits( - self, cross_org_referrer_limits: dict[str, dict[str, int]] - ) -> None: - for referrer, limits in cross_org_referrer_limits.items(): - concurrent_limit = limits.get("concurrent_limit", None) - max_threads = limits.get("max_threads", None) - if not isinstance(max_threads, int): - raise ValueError(f"max_threads is required for {referrer} and must be an int") - if not isinstance(concurrent_limit, int): - raise ValueError(f"concurrent_limit is required for {referrer} and must be an int") - - def __init__( - self, - storage_key: ResourceIdentifier, - **kwargs: Any, - ) -> None: - super().__init__( - storage_key, - **kwargs, - ) - self._registered_cross_org_referrers = cast( - "dict[str, dict[str, int]]", kwargs.get("cross_org_referrer_limits", {}) - ) - self._validate_cross_org_referrer_limits(self._registered_cross_org_referrers) - - def _get_max_threads(self, referrer: str) -> int: - if not self._referrer_is_registered(referrer): - return _UNREGISTERED_REFERRER_MAX_THREADS - return int(self._registered_cross_org_referrers[referrer]["max_threads"]) - - def _get_concurrent_limit(self, referrer: str) -> int: - if not self._referrer_is_registered(referrer): - return _UNREGISTERED_REFERRER_CONCURRENT_QUERIES - concurrent_override = int( - self.get_config_value("referrer_concurrent_override", {"referrer": referrer}) - ) - return ( - concurrent_override - if concurrent_override != -1 - else int(self._registered_cross_org_referrers[referrer]["concurrent_limit"]) - ) - - def _referrer_is_registered(self, referrer: str) -> bool: - return referrer in self._registered_cross_org_referrers - - def _get_quota_allowance( - self, tenant_ids: dict[str, str | int], query_id: str - ) -> QuotaAllowance: - referrer = str(tenant_ids.get("referrer", "no_referrer")) - if not self._referrer_is_registered(referrer) and not self.is_cross_org_query(tenant_ids): - # This is not a cross org query and the referrer is not registered. This is outside the responsibility of this policy - return QuotaAllowance( - can_run=True, - max_threads=self.max_threads, - explanation={"reason": "pass_through"}, - is_throttled=False, - throttle_threshold=MAX_THRESHOLD, - rejection_threshold=MAX_THRESHOLD, - quota_used=0, - quota_unit=NO_UNITS, - suggestion=NO_SUGGESTION, - ) - - concurrent_limit = self._get_concurrent_limit(referrer) - rate_limit_params = RateLimitParameters( - self.rate_limit_name, referrer, None, concurrent_limit - ) - rate_limit_stats, can_run, explanation = self._is_within_rate_limit( - query_id, - rate_limit_params, - ) - decision_explanation: dict[str, JsonSerializable] = {"reason": explanation} - if not self._referrer_is_registered(referrer): - decision_explanation["cross_org_query"] = ( - f"This referrer is not registered for the current storage {self._resource_identifier.value}, if you want to increase its limits, register it in the yaml of the CrossOrgQueryAllocationPolicy" - ) - - return QuotaAllowance( - can_run=can_run, - max_threads=self._get_max_threads(referrer) if can_run else 0, - explanation=decision_explanation, - is_throttled=False, - throttle_threshold=typing.cast(int, rate_limit_params.concurrent_limit), - rejection_threshold=typing.cast(int, rate_limit_params.concurrent_limit), - quota_used=rate_limit_stats.concurrent, - quota_unit=QUOTA_UNIT, - suggestion=NO_SUGGESTION if can_run else SUGGESTION, - ) - - def _update_quota_balance( - self, - tenant_ids: dict[str, str | int], - query_id: str, - result_or_error: QueryResultOrError, - ) -> None: - referrer = str(tenant_ids.get("referrer", "no_referrer")) - if not self._referrer_is_registered(referrer) and not self.is_cross_org_query(tenant_ids): - return - rate_limit_params = RateLimitParameters( - self.rate_limit_name, - referrer, - None, - # limit number does not matter for ending a query so I just picked 22 - 22, - ) - self._end_query(query_id, rate_limit_params, result_or_error) diff --git a/tests/query/allocation_policies/test_allocation_policy_base.py b/tests/query/allocation_policies/test_allocation_policy_base.py index ff3edbf94f..63b1ca7681 100644 --- a/tests/query/allocation_policies/test_allocation_policy_base.py +++ b/tests/query/allocation_policies/test_allocation_policy_base.py @@ -28,7 +28,6 @@ from snuba.query.allocation_policies.concurrent_rate_limit import ( ConcurrentRateLimitAllocationPolicy, ) -from snuba.query.allocation_policies.cross_org import CrossOrgQueryAllocationPolicy from snuba.query.allocation_policies.per_referrer import ReferrerGuardRailPolicy from snuba.utils.metrics.backends.testing import get_recorded_metric_calls from tests.configs.component_config import ( @@ -508,7 +507,6 @@ def get_all_allocation_policy_classes(self) -> list[type[AllocationPolicy]]: ReferrerGuardRailPolicy, ConcurrentRateLimitAllocationPolicy, BytesScannedRejectingPolicy, - CrossOrgQueryAllocationPolicy, BytesScannedWindowAllocationPolicy, ] diff --git a/tests/query/allocation_policies/test_attachment.py b/tests/query/allocation_policies/test_attachment.py index 7c671059dc..62536a72cd 100644 --- a/tests/query/allocation_policies/test_attachment.py +++ b/tests/query/allocation_policies/test_attachment.py @@ -16,11 +16,6 @@ {"name": "BytesScannedRejectingPolicy", "is_enforced": 0}, {"name": "BytesScannedWindowAllocationPolicy", "is_enforced": 0}, {"name": "ConcurrentRateLimitAllocationPolicy", "is_enforced": 0}, - { - "name": "CrossOrgQueryAllocationPolicy", - "is_enforced": 0, - "cross_org_referrer_limits": {"some.referrer": {"max_threads": 2, "concurrent_limit": 4}}, - }, {"name": "DeleteConcurrentRateLimitAllocationPolicy", "is_enforced": 0}, {"name": "PassthroughPolicy", "is_enforced": 0}, {"name": "ReferrerGuardRailPolicy", "is_enforced": 0}, diff --git a/tests/query/allocation_policies/test_cross_org_policy.py b/tests/query/allocation_policies/test_cross_org_policy.py deleted file mode 100644 index b311e9aeac..0000000000 --- a/tests/query/allocation_policies/test_cross_org_policy.py +++ /dev/null @@ -1,170 +0,0 @@ -import pytest - -from snuba.query.allocation_policies import QueryResultOrError -from snuba.query.allocation_policies.cross_org import CrossOrgQueryAllocationPolicy -from snuba.web import QueryResult -from tests.configs.component_config import set_component_config - -_RESULT_SUCCESS = QueryResultOrError( - QueryResult( - result={"profile": {"bytes": 42069}}, - extra={"stats": {}, "sql": "", "experiments": {}}, - ), - error=None, -) - - -class TestCrossOrgQueryAllocationPolicy: - @pytest.mark.redis_db - def test_policy_pass_basic(self): - policy = CrossOrgQueryAllocationPolicy.from_kwargs( - **{ - "storage_key": "metrics_counters", - "cross_org_referrer_limits": { - "statistical_detectors": { - "concurrent_limit": 1, - "max_threads": 1, - }, - }, - } - ) - unimportant_allowance = policy.get_quota_allowance( - tenant_ids={"referrer": "some_referrer"}, query_id="1" - ) - assert unimportant_allowance.can_run is True - assert unimportant_allowance.max_threads == 10 - assert unimportant_allowance.explanation == { - "reason": "pass_through", - "storage_key": "metrics_counters", - } - cross_org_allowance = policy.get_quota_allowance( - tenant_ids={"referrer": "statistical_detectors"}, query_id="2" - ) - assert cross_org_allowance.can_run is True - assert cross_org_allowance.max_threads == 1 - - quota_allowance = policy.get_quota_allowance( - tenant_ids={"referrer": "statistical_detectors"}, query_id="3" - ) - assert not quota_allowance.can_run and quota_allowance.max_threads == 0 - policy.update_quota_balance( - tenant_ids={"referrer": "statistical_detectors"}, - query_id="2", - result_or_error=_RESULT_SUCCESS, - ) - assert policy.get_quota_allowance( - tenant_ids={"referrer": "statistical_detectors"}, query_id="3" - ).can_run - - @pytest.mark.parametrize( - "config", - [ - { - "storage_key": "metrics_counters", - "cross_org_referrer_limits": { - "statistical_detectors": { - "max_threads": 1, - }, - }, - }, - { - "storage_key": "metrics_counters", - "cross_org_referrer_limits": { - "statistical_detectors": { - "concurrent_limit": 2, - }, - }, - }, - { - "storage_key": "metrics_counters", - "cross_org_referrer_limits": { - "statistical_detectors": {"max_threads": {}, "concurrent_limit": 2}, - }, - }, - ], - ) - def test_bad_config(self, config) -> None: - with pytest.raises(ValueError): - CrossOrgQueryAllocationPolicy.from_kwargs(**config) - - @pytest.mark.redis_db - def test_override(self): - policy = CrossOrgQueryAllocationPolicy.from_kwargs( - **{ - "storage_key": "metrics_counters", - "cross_org_referrer_limits": { - "statistical_detectors": { - "concurrent_limit": 1, - "max_threads": 1, - }, - }, - } - ) - set_component_config( - policy, - "referrer_concurrent_override", - 0, - {"referrer": "statistical_detectors"}, - ) - - quota_allowance = policy.get_quota_allowance( - tenant_ids={"referrer": "statistical_detectors"}, query_id="2" - ) - assert not quota_allowance.can_run and quota_allowance.max_threads == 0 - - @pytest.mark.redis_db - def test_respects_max_threads_config(self) -> None: - policy = CrossOrgQueryAllocationPolicy.from_kwargs( - **{ - "storage_key": "metrics_counters", - "cross_org_referrer_limits": { - "statistical_detectors": { - "concurrent_limit": 10, - "max_threads": 4, - }, - }, - } - ) - set_component_config(policy, "max_threads", 7) - assert ( - policy.get_quota_allowance( - tenant_ids={"referrer": "some_referrer"}, query_id="1" - ).max_threads - == 7 - ) - assert ( - policy.get_quota_allowance( - tenant_ids={"referrer": "statistical_detectors"}, query_id="2" - ).max_threads - == 4 - ) - - @pytest.mark.redis_db - def test_throttle_cross_org_query_with_unregistered_referrer(self): - policy = CrossOrgQueryAllocationPolicy.from_kwargs( - **{ - "storage_key": "metrics_counters", - "cross_org_referrer_limits": { - "statistical_detectors": { - "concurrent_limit": 1, - "max_threads": 1, - }, - }, - } - ) - allowance = policy.get_quota_allowance( - tenant_ids={"referrer": "unregistered", "cross_org_query": 1}, - query_id="1", - ) - assert allowance.can_run is True - assert allowance.max_threads == 1 - - allowance = policy.get_quota_allowance( - tenant_ids={"referrer": "unregistered", "cross_org_query": 1}, - query_id="2", - ) - assert not allowance.can_run - assert ( - allowance.explanation["cross_org_query"] - == "This referrer is not registered for the current storage metrics_counters, if you want to increase its limits, register it in the yaml of the CrossOrgQueryAllocationPolicy" - )