ref(options): migrate rate-limit runtime config to sentry-options

sentry-options/schemas/snuba/schema.json

@@ -258,6 +258,89 @@
       },
       "default": {},
       "description": "Dict mapping an outcomes-based routing strategy class name to the minimum query time range in seconds for which outcomes are queried. Strategies with no entry use 14400 (4 hours). Migrated from per-strategy runtime config <ClassName>.min_timerange_to_query_outcomes."
+    },
+    "project_quota_time_percentage": {
+      "type": "number",
+      "default": 1.0,
+      "description": "Fraction of the counter window a project may spend on replacements before it is reported as exceeding the time limit."
+    },
+    "counter_window_size_minutes": {
+      "type": "integer",
+      "default": 10,
+      "description": "Size in minutes of the rolling window the replacement bucket timer uses to track per-project processing time."
+    },
+    "allows_skipping_single_project_replacements": {
+      "type": "boolean",
+      "default": false,
+      "description": "When true, a single project that exceeds the replacement time limit can be skipped (otherwise only multi-project groups are skipped)."
+    },
+    "mem_rate_limit_per_sec": {
+      "type": "object",
+      "additionalProperties": {
+        "type": "number"
+      },
+      "default": {},
+      "description": "Dict mapping in-process rate-limit bucket name to a maximum operations-per-second. Buckets with no entry are unlimited (rate limiter off). Migrated from per-bucket runtime config mem_rate_limit_per_sec_<bucket>."
+    },
+    "bypass_rate_limit": {
+      "type": "integer",
+      "default": 0,
+      "description": "When set to 1, all Redis-backed rate limits are bypassed (no limiting applied). Migrated from runtime config bypass_rate_limit."
+    },
+    "rate_history_sec": {
+      "type": "integer",
+      "default": 3600,
+      "description": "Number of seconds the rate limiter keeps per-request timestamps in its Redis sorted set. Migrated from runtime config rate_history_sec."
+    },
+    "rate_limit_shard_factor": {
+      "type": "integer",
+      "default": 1,
+      "description": "Number of shards each rate-limit Redis set is split into. Increasing it multiplies the number of Redis keys and reduces the size of each set. Migrated from runtime config rate_limit_shard_factor."
+    },
+    "mandatory_condition_enforce": {
+      "type": "boolean",
+      "default": false,
+      "description": "When true, queries missing mandatory condition columns raise an assertion; otherwise the omission is only logged."
+    },
+    "skip_final_subscriptions_projects": {
+      "type": "string",
+      "default": "[]",
+      "description": "Bracketed comma-separated list of project ids for which subscription queries skip the FINAL keyword."
+    },
+    "post_replacement_consistency_projects_denylist": {
+      "type": "string",
+      "default": "[]",
+      "description": "Bracketed comma-separated list of project ids that are forced to FINAL after a replacement."
+    },
+    "max_group_ids_exclude": {
+      "type": "integer",
+      "default": 256,
+      "description": "Maximum number of group ids excluded from a query before it falls back to FINAL instead of an exclusion set."
+    },
+    "skip_seen_offsets": {
+      "type": "boolean",
+      "default": false,
+      "description": "When true, the replacer skips replacement messages whose offset has already been seen."
+    },
+    "consumer_groups_to_reset_offset_check": {
+      "type": "string",
+      "default": "[]",
+      "description": "Bracketed comma-separated list of consumer groups whose replacer offset-seen check should be reset."
+    },
+    "write_node_replacements_global": {
+      "type": "number",
+      "default": 1.0,
+      "description": "Probability [0,1] that a replacement is written to every storage node rather than only the distributed table."
+    },
+    "replacements_bypass_projects": {
+      "type": "string",
+      "default": "[]",
+      "description": "JSON array of project ids for which error replacements are skipped."
+    },
+    "replacements_expiry_window_minutes": {
+      "type": "integer",
+      "default": 5,
+      "description": "Window in minutes for which a project that recently received replacements is kept in the auto-replacements bypass set. Migrated from runtime config replacements_expiry_window_minutes."
     }
   }
 }

snuba/query/processors/physical/conditions_enforcer.py

@@ -6,7 +6,7 @@
 from snuba.query.processors.condition_checkers import ConditionChecker
 from snuba.query.processors.physical import ClickhouseQueryProcessor
 from snuba.query.query_settings import QuerySettings
-from snuba.state import get_config
+from snuba.state.sentry_options import get_option
 
 logger = logging.getLogger(__name__)
 
@@ -47,7 +47,7 @@ def inspect_expression(condition: Expression) -> None:
             inspect_expression(prewhere)
 
         missing_ids = {checker.get_id() for checker in missing_checkers}
-        if get_config("mandatory_condition_enforce", 0):
+        if get_option("mandatory_condition_enforce", False):
             assert not missing_checkers, (
                 f"Missing mandatory columns in query. Missing {missing_ids}"
             )

snuba/query/processors/physical/replaced_groups.py

@@ -14,7 +14,7 @@
 from snuba.query.query_settings import QuerySettings, SubscriptionQuerySettings
 from snuba.replacers.projects_query_flags import ProjectsQueryFlags
 from snuba.replacers.replacer_processor import ReplacerState
-from snuba.state import get_config
+from snuba.state.sentry_options import get_option
 from snuba.utils.metrics.wrapper import MetricsWrapper
 
 metrics = MetricsWrapper(environment.metrics, "processors.replaced_groups")
@@ -52,7 +52,7 @@ def process_query(self, query: Query, query_settings: QuerySettings) -> None:
             return
 
         for no_final_subscriptions_project in (
-            get_config("skip_final_subscriptions_projects") or "[]"
+            get_option("skip_final_subscriptions_projects", "[]")
         )[1:-1].split(","):
             if (
                 no_final_subscriptions_project
@@ -64,7 +64,7 @@ def process_query(self, query: Query, query_settings: QuerySettings) -> None:
                 return
 
         for denied_project_id_string in (
-            get_config("post_replacement_consistency_projects_denylist") or "[]"
+            get_option("post_replacement_consistency_projects_denylist", "[]")
         )[1:-1].split(","):
             if denied_project_id_string and int(denied_project_id_string) in project_ids:
                 metrics.increment(name=CONSISTENCY_DENYLIST_METRIC)
@@ -96,11 +96,9 @@ def process_query(self, query: Query, query_settings: QuerySettings) -> None:
         elif flags.group_ids_to_exclude:
             # If the number of groups to exclude exceeds our limit, the query
             # should just use final instead of the exclusion set.
-            max_group_ids_exclude = get_config(
-                "max_group_ids_exclude",
-                settings.REPLACER_MAX_GROUP_IDS_TO_EXCLUDE,
+            max_group_ids_exclude = get_option(
+                "max_group_ids_exclude", settings.REPLACER_MAX_GROUP_IDS_TO_EXCLUDE
             )
-            assert isinstance(max_group_ids_exclude, int)
             groups_to_exclude = self._groups_to_exclude(query, flags.group_ids_to_exclude)
             if (
                 len(flags.group_ids_to_exclude) > 2 * max_group_ids_exclude

snuba/replacer.py

@@ -42,7 +42,7 @@
     ReplacementMessage,
     ReplacementMessageMetadata,
 )
-from snuba.state import get_int_config, get_str_config
+from snuba.state.sentry_options import get_option
 from snuba.utils.bucket_timer import Counter
 from snuba.utils.metrics import MetricsBackend
 from snuba.utils.rate_limiter import RateLimiter
@@ -416,7 +416,7 @@ def process_message(
                     "offset": metadata.offset,
                 },
             )
-            if get_int_config("skip_seen_offsets"):
+            if get_option("skip_seen_offsets", False):
                 return None
         seq_message = json.loads(message.payload.value)
         [version, action_type, data] = seq_message
@@ -522,7 +522,7 @@ def _reset_offset_check(self, key: str) -> None:
         temporarily, then cleared once relevant consumers restart.
         """
         # expected format is "[consumer_group1,consumer_group2,..]"
-        consumer_groups = (get_str_config(RESET_CHECK_CONFIG) or "[]")[1:-1].split(",")
+        consumer_groups = (get_option(RESET_CHECK_CONFIG, "[]"))[1:-1].split(",")
         if self.__consumer_group in consumer_groups:
             self.__last_offset_processed_per_partition[key] = -1
             redis_client.delete(key)

snuba/replacers/errors_replacer.py

@@ -47,7 +47,7 @@
     ReplacerProcessor,
     ReplacerState,
 )
-from snuba.state import get_config, get_float_config
+from snuba.state.sentry_options import get_option
 from snuba.utils.metrics.wrapper import MetricsWrapper
 
 """
@@ -101,8 +101,7 @@ def get_project_id(self) -> int:
         raise NotImplementedError()
 
     def should_write_every_node(self) -> bool:
-        write_node_replacement_setting = get_float_config("write_node_replacements_global", 1.0)
-        assert isinstance(write_node_replacement_setting, float)
+        write_node_replacement_setting = get_option("write_node_replacements_global", 1.0)
         return random.random() < write_node_replacement_setting
 
 
@@ -180,7 +179,7 @@ def process_message(self, message: ReplacementMessage[Mapping[str, Any]]) -> Rep
             raise InvalidMessageType(f"Invalid message type: {type_}")
 
         if processed is not None:
-            manual_bypass_projects = get_config("replacements_bypass_projects", "[]")
+            manual_bypass_projects = get_option("replacements_bypass_projects", "[]")
             auto_bypass_projects = list(
                 get_config_auto_replacements_bypass_projects(datetime.now()).keys()
             )

snuba/replacers/projects_query_flags.py

@@ -14,7 +14,7 @@
 from snuba.processor import ReplacementType
 from snuba.redis import RedisClientKey, get_redis_client
 from snuba.replacers.replacer_processor import ReplacerState
-from snuba.state import get_config
+from snuba.state.sentry_options import get_option
 
 redis_client = get_redis_client(RedisClientKey.REPLACEMENTS_STORE)
 
@@ -83,11 +83,9 @@ def set_project_exclude_groups(
             # the redis key size limit is defined as 2 times the clickhouse query size
             # limit. there is an explicit check in the query processor for the same
             # limit
-            max_group_ids_exclude = get_config(
-                "max_group_ids_exclude",
-                settings.REPLACER_MAX_GROUP_IDS_TO_EXCLUDE,
+            max_group_ids_exclude = get_option(
+                "max_group_ids_exclude", settings.REPLACER_MAX_GROUP_IDS_TO_EXCLUDE
             )
-            assert isinstance(max_group_ids_exclude, int)
 
             group_id_data: MutableMapping[str | bytes, bytes | float | int | str] = {}
             for group_id in group_ids:

snuba/replacers/replacements_and_expiry.py

@@ -2,13 +2,12 @@
 
 import logging
 import time
-import typing
 from collections.abc import Mapping, Sequence
 from datetime import datetime, timedelta
 
 from snuba import environment
 from snuba.redis import RedisClientKey, get_redis_client
-from snuba.state import get_int_config
+from snuba.state.sentry_options import get_option
 from snuba.utils.metrics.wrapper import MetricsWrapper
 
 logger = logging.getLogger(__name__)
@@ -29,9 +28,7 @@ def set_config_auto_replacements_bypass_projects(
     try:
         projects_within_expiry = get_config_auto_replacements_bypass_projects(curr_time)
         start = time.time()
-        expiry_window = typing.cast(
-            int, get_int_config(key=REPLACEMENTS_EXPIRY_WINDOW_MINUTES_KEY, default=5)
-        )
+        expiry_window = get_option(REPLACEMENTS_EXPIRY_WINDOW_MINUTES_KEY, 5)
         with redis_client.pipeline() as pipeline:
             for project_id in new_project_ids:
                 if project_id not in projects_within_expiry:

snuba/state/rate_limit.py

@@ -17,6 +17,7 @@
 from snuba import environment, state
 from snuba.redis import RedisClientKey, get_redis_client
 from snuba.state import get_configs, set_config
+from snuba.state.sentry_options import get_option
 from snuba.utils.metrics.wrapper import MetricsWrapper
 from snuba.utils.serializable_exception import SerializableException
 
@@ -347,24 +348,14 @@ def rate_limit(
             # will raise RateLimitExceeded if the rate limit is exceeded
 
     """
-    (
-        bypass_rate_limit,
-        rate_history_s,
-        rate_limit_shard_factor,
-    ) = state.get_configs(
-        [
-            # bool (0/1) flag to disable rate limits altogether
-            ("bypass_rate_limit", 0),
-            # number of seconds the timestamps are kept
-            ("rate_history_sec", 3600),
-            # number of shards that each redis set is supposed to have.
-            # increasing this value multiplies the number of redis keys by that
-            # factor, and (on average) reduces the size of each redis set
-            ("rate_limit_shard_factor", 1),
-        ]
-    )
-    assert isinstance(rate_history_s, int)
-    assert isinstance(rate_limit_shard_factor, int)
+    # bool (0/1) flag to disable rate limits altogether
+    bypass_rate_limit = get_option("bypass_rate_limit", 0)
+    # number of seconds the timestamps are kept
+    rate_history_s = get_option("rate_history_sec", 3600)
+    # number of shards that each redis set is supposed to have. increasing this
+    # value multiplies the number of redis keys by that factor, and (on average)
+    # reduces the size of each redis set
+    rate_limit_shard_factor = get_option("rate_limit_shard_factor", 1)
     assert rate_limit_shard_factor > 0
 
     if bypass_rate_limit == 1:

snuba/utils/bucket_timer.py

@@ -1,12 +1,11 @@
 from __future__ import annotations
 
-import typing
 from collections import defaultdict
 from collections.abc import MutableMapping
 from datetime import datetime, timedelta
 
-from snuba import environment, state
-from snuba.state import get_int_config
+from snuba import environment
+from snuba.state.sentry_options import get_option
 from snuba.utils.metrics.wrapper import MetricsWrapper
 
 metrics = MetricsWrapper(environment.metrics, "bucket_timer")
@@ -37,11 +36,8 @@ def __init__(self, consumer_group: str) -> None:
         self.consumer_group: str = consumer_group
         self.buckets: Buckets = {}
 
-        percentage = state.get_config("project_quota_time_percentage", 1.0)
-        assert isinstance(percentage, float)
-        counter_window_size_minutes = typing.cast(
-            int, get_int_config(key="counter_window_size_minutes", default=10)
-        )
+        percentage = get_option("project_quota_time_percentage", 1.0)
+        counter_window_size_minutes = get_option("counter_window_size_minutes", 10)
         self.counter_window_size = timedelta(minutes=counter_window_size_minutes)
         self.limit = self.counter_window_size * percentage
 
@@ -92,7 +88,7 @@ def get_projects_exceeding_limit(self) -> list[int]:
         for project_id, total_processing_time in project_groups.items():
             if total_processing_time > self.limit and (
                 len(project_groups) > 1
-                or get_int_config("allows_skipping_single_project_replacements", default=0)
+                or get_option("allows_skipping_single_project_replacements", False)
             ):
                 projects_exceeding_time_limit.append(project_id)
 

snuba/utils/rate_limiter.py

@@ -5,9 +5,12 @@
 from threading import Lock
 from typing import Any
 
-from snuba import state
+from snuba.state.sentry_options import get_mapped_option
 
-RATE_LIMIT_PER_SEC_KEY_PREFIX = "mem_rate_limit_per_sec_"
+# sentry-options dict option whose keys are rate-limit bucket names and whose
+# values are the per-bucket max operations-per-second. Migrated from the
+# per-bucket runtime config keys "mem_rate_limit_per_sec_<bucket>".
+RATE_LIMIT_PER_SEC_OPTION = "mem_rate_limit_per_sec"
 
 
 class RateLimitResult(Enum):
@@ -39,7 +42,7 @@ def __init__(self, bucket: str, max_rate_per_sec: float | None = None) -> None:
 
     def __enter__(self) -> tuple[RateLimitResult, int]:
         limit = (
-            state.get_config(f"{RATE_LIMIT_PER_SEC_KEY_PREFIX}{self.__bucket}", None)
+            (get_mapped_option(RATE_LIMIT_PER_SEC_OPTION, self.__bucket, 0.0))
             if not self.__max_rate_per_sec
             else self.__max_rate_per_sec
         )