diff --git a/fixtures/stubs-for-mypy/rediscluster/nodemanager.pyi b/fixtures/stubs-for-mypy/rediscluster/nodemanager.pyi new file mode 100644 index 000000000000..970b0f04a8b8 --- /dev/null +++ b/fixtures/stubs-for-mypy/rediscluster/nodemanager.pyi @@ -0,0 +1,5 @@ +from typing import Any + +class NodeManager: + def __init__(self, startup_nodes: list[dict[str, Any]], **kwargs: object) -> None: ... + def keyslot(self, key: str | bytes) -> int: ... diff --git a/src/sentry/conf/server.py b/src/sentry/conf/server.py index d6ec5166b331..d32beb986e21 100644 --- a/src/sentry/conf/server.py +++ b/src/sentry/conf/server.py @@ -1139,6 +1139,10 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str: "task": "uptime:sentry.uptime.tasks.config_drift_dispatcher", "schedule": crontab("0", "*/1", "*", "*", "*"), }, + "uptime-config-sentinel-checker": { + "task": "uptime:sentry.uptime.tasks.check_config_sentinels", + "schedule": crontab("*/1", "*", "*", "*", "*"), + }, "poll_tempest": { "task": "tempest:sentry.tempest.tasks.poll_tempest", "schedule": crontab("*/1", "*", "*", "*", "*"), diff --git a/src/sentry/options/defaults.py b/src/sentry/options/defaults.py index 3b45356c1b3d..dea043efe401 100644 --- a/src/sentry/options/defaults.py +++ b/src/sentry/options/defaults.py @@ -3725,6 +3725,15 @@ flags=FLAG_AUTOMATOR_MODIFIABLE, ) +# Kill switch for sentinel-triggered whole-store repair. Defaults to True (repair off) so the +# option is never left permanently True with no way to remove it. +register( + "uptime.config-drift.sentinel-repair-disabled", + type=Bool, + default=True, + flags=FLAG_AUTOMATOR_MODIFIABLE, +) + # Controls whether uptime monitoring automatically detects hostnames from error events. register( "uptime.automatic-hostname-detection", diff --git a/src/sentry/uptime/config_drift.py b/src/sentry/uptime/config_drift.py index d15faecfc0c0..24e72862f817 100644 --- a/src/sentry/uptime/config_drift.py +++ b/src/sentry/uptime/config_drift.py @@ -20,6 +20,8 @@ SUBSCRIPTION_ID_PREFIX_BUCKETS = 256 # Bounds the IN list when a partition holds many configs. IN_CHUNK_SIZE = 1000 +# Refreshed by every run of a pass; if it lapses, the pass restarts from the first missing id. +REPAIR_CURSOR_TTL = 3600 @dataclass(frozen=True) @@ -78,7 +80,7 @@ def _live_region_rows() -> QuerySet[UptimeSubscriptionRegion]: def find_missing_configs(store: ConfigStore, subscription_id_prefix: str) -> DriftResult: - cluster = redis.redis_clusters.get_binary(store.cluster) + cluster = redis.redis_clusters.get(store.cluster) pipe = cluster.pipeline() subscription_ids: list[str] = [] # One row per (subscription, region); slugs sharing a store must be checked once. @@ -103,10 +105,8 @@ def find_missing_configs(store: ConfigStore, subscription_id_prefix: str) -> Dri def find_orphaned_configs(store: ConfigStore, partition: int) -> DriftResult: - cluster = redis.redis_clusters.get_binary(store.cluster) - stored = { - field.decode() for field in cluster.hkeys(get_config_key(store.key_prefix, partition)) - } + cluster = redis.redis_clusters.get(store.cluster) + stored = set(cluster.hkeys(get_config_key(store.key_prefix, partition))) live: set[str | None] = set() for chunk in batched(stored, IN_CHUNK_SIZE): live.update( @@ -118,3 +118,53 @@ def find_orphaned_configs(store: ConfigStore, partition: int) -> DriftResult: .values_list("uptime_subscription__subscription_id", flat=True) ) return DriftResult(checked=len(stored), drifted_ids=frozenset(stored - live)) + + +def find_missing_configs_for_store(store: ConfigStore) -> DriftResult: + """ + Whole-store form of find_missing_configs: reads every partition's field names in one + pipeline instead of probing one id at a time, since here every row is checked anyway. + """ + # Postgres first: a row ACTIVE now had its config written before now, so a config + # published between the two reads can't show up as missing. + rows = ( + _active_region_rows() + .filter(region_slug__in=store.region_slugs) + .values_list("uptime_subscription__subscription_id", flat=True) + ) + live = {subscription_id for subscription_id in rows if subscription_id is not None} + cluster = redis.redis_clusters.get(store.cluster) + pipe = cluster.pipeline() + for partition in range(settings.UPTIME_CONFIG_PARTITIONS): + pipe.hkeys(get_config_key(store.key_prefix, partition)) + stored = {field for fields in pipe.execute() for field in fields} + return DriftResult(checked=len(live), drifted_ids=frozenset(live - stored)) + + +def get_sentinel_key(key_prefix: str, partition: int) -> str: + # Redis hash tag: lands in the config key's slot, so a node that loses it loses this too. + return f"{{{get_config_key(key_prefix, partition)}}}:sentinel" + + +def get_repair_cursor_key(key_prefix: str) -> str: + return f"{key_prefix}uptime:configs:repair-cursor" + + +def clear_repair_cursor(store: ConfigStore) -> None: + redis.redis_clusters.get(store.cluster).delete(get_repair_cursor_key(store.key_prefix)) + + +def find_missing_sentinels(store: ConfigStore) -> list[int]: + cluster = redis.redis_clusters.get(store.cluster) + pipe = cluster.pipeline() + for partition in range(settings.UPTIME_CONFIG_PARTITIONS): + pipe.exists(get_sentinel_key(store.key_prefix, partition)) + return [partition for partition, present in enumerate(pipe.execute()) if not present] + + +def write_sentinels(store: ConfigStore) -> None: + cluster = redis.redis_clusters.get(store.cluster) + pipe = cluster.pipeline() + for partition in range(settings.UPTIME_CONFIG_PARTITIONS): + pipe.set(get_sentinel_key(store.key_prefix, partition), "1") + pipe.execute() diff --git a/src/sentry/uptime/subscriptions/tasks.py b/src/sentry/uptime/subscriptions/tasks.py index d3a5e95a82a8..e6c40d8ac9cc 100644 --- a/src/sentry/uptime/subscriptions/tasks.py +++ b/src/sentry/uptime/subscriptions/tasks.py @@ -15,13 +15,19 @@ from sentry.tasks.base import instrumented_task from sentry.taskworker.namespaces import uptime_tasks from sentry.uptime.config_drift import ( + REPAIR_CURSOR_TTL, SUBSCRIPTION_ID_PREFIX_BUCKETS, SWEEP_RUN_INTERVAL, ConfigStore, + clear_repair_cursor, find_missing_configs, + find_missing_configs_for_store, + find_missing_sentinels, find_orphaned_configs, get_config_stores, + get_repair_cursor_key, sweep_slice, + write_sentinels, ) from sentry.uptime.config_producer import produce_config, produce_config_removal from sentry.uptime.models import ( @@ -30,7 +36,7 @@ UptimeSubscriptionRegion, ) from sentry.uptime.types import CheckConfig -from sentry.utils import metrics +from sentry.utils import metrics, redis from sentry.utils.audit import create_system_audit_entry from sentry.utils.query import RangeQuerySetWrapper @@ -371,6 +377,88 @@ def check_orphaned_configs(cluster: str, key_prefix: str, partition: int, **kwar ) +@instrumented_task( + name="sentry.uptime.tasks.check_config_sentinels", + namespace=uptime_tasks, + processing_deadline_duration=60, +) +def check_config_sentinels(**kwargs): + """ + Checks each config store's partition sentinels every minute; a missing sentinel means the + store lost data, so its whole-store comparison is queued unless sentinel repair is disabled. + """ + if not options.get("uptime.config-drift.enabled"): + return + + for store in get_config_stores(): + # The store that failed is likely the one being lost; keep checking the others. + try: + missing = find_missing_sentinels(store) + except Exception: + logger.exception("uptime.config_drift.sentinel_check_failed") + continue + metrics.gauge( + "uptime.config_drift.sentinel_missing", + len(missing), + tags={"cluster": store.cluster}, + sample_rate=1.0, + ) + if not missing: + # No pass runs while every sentinel is present, so a cursor here was left by an + # overlapping run or an old worker and must not carry over into the next loss. + try: + clear_repair_cursor(store) + except Exception: + logger.exception("uptime.config_drift.repair_cursor_clear_failed") + if missing and not options.get("uptime.config-drift.sentinel-repair-disabled"): + logger.warning( + "uptime.config_drift.sentinel_missing", + extra={"cluster": store.cluster, "count": len(missing), "partitions": missing}, + ) + repair_config_store.delay(cluster=store.cluster, key_prefix=store.key_prefix) + + +@instrumented_task( + name="sentry.uptime.tasks.repair_config_store", + namespace=uptime_tasks, + processing_deadline_duration=60, + expires=60, +) +def repair_config_store(cluster: str, key_prefix: str, **kwargs): + """ + One step of a pass over the store's missing configs after a sentinel went missing. + """ + store = _find_store(cluster, key_prefix) + if store is None: + return + + client = redis.redis_clusters.get(store.cluster) + cursor_key = get_repair_cursor_key(store.key_prefix) + cursor = client.get(cursor_key) or "" + result = find_missing_configs_for_store(store) + # Each missing id is queued once per pass, so ids that can't be published neither hold + # the sentinels back nor crowd out the ids after them. + pending = sorted(i for i in result.drifted_ids if i > cursor) + batch = pending[:CONFIG_REPAIR_MAX_TASKS] + logger.info( + "uptime.config_drift.store_repair", + extra={ + "cluster": store.cluster, + "checked": result.checked, + "missing": len(result.drifted_ids), + "pending": len(pending), + }, + ) + if batch: + repair_missing_configs(store, batch, limit=CONFIG_REPAIR_MAX_TASKS) + if len(pending) <= CONFIG_REPAIR_MAX_TASKS: + # Cleared first, so a failed sentinel write restarts the pass rather than skipping ids. + client.delete(cursor_key) + write_sentinels(store) + else: + client.set(cursor_key, batch[-1], ex=REPAIR_CURSOR_TTL) + + def repair_missing_configs( store: ConfigStore, subscription_ids: Collection[str], *, limit: int = CONFIG_REPAIR_MAX_TASKS ) -> int: diff --git a/tests/sentry/uptime/test_config_drift.py b/tests/sentry/uptime/test_config_drift.py index 144fe9146098..3557fcadec85 100644 --- a/tests/sentry/uptime/test_config_drift.py +++ b/tests/sentry/uptime/test_config_drift.py @@ -8,11 +8,13 @@ from django.db import connections, router from django.test import override_settings from django.test.utils import CaptureQueriesContext +from rediscluster.nodemanager import NodeManager from sentry.conf.types.uptime import UptimeRegionConfig from sentry.testutils.cases import UptimeTestCase from sentry.testutils.helpers import override_options from sentry.testutils.helpers.datetime import freeze_time +from sentry.uptime.config_drift import get_repair_cursor_key, get_sentinel_key from sentry.uptime.config_producer import ( get_config_key, get_partition_from_subscription_id, @@ -21,9 +23,11 @@ from sentry.uptime.models import UptimeSubscription, UptimeSubscriptionRegion from sentry.uptime.subscriptions import tasks from sentry.uptime.subscriptions.tasks import ( + check_config_sentinels, check_missing_configs, check_orphaned_configs, config_drift_dispatcher, + repair_config_store, update_remote_uptime_subscription, uptime_subscription_to_check_config, ) @@ -342,3 +346,129 @@ def test_tasks_write_nothing_to_postgres(self) -> None: if q["sql"].startswith(("INSERT", "UPDATE", "DELETE")) ] assert writes == [] + + +def test_sentinel_shares_slot_with_partition_hash() -> None: + # Offline slot math; the node is never contacted. + keyslot = NodeManager(startup_nodes=[{"host": "localhost", "port": 1}]).keyslot + for key_prefix in ("", "a"): + for partition in range(128): + assert keyslot(get_sentinel_key(key_prefix, partition)) == keyslot( + get_config_key(key_prefix, partition) + ) + + +@override_settings(UPTIME_REGIONS=REGIONS) +class CheckConfigSentinelsTest(ConfigPusherTestMixin): + def setUp(self) -> None: + super().setUp() + self.enterContext(override_options({"uptime.config-drift.enabled": True})) + + def test_sentinel_repair_disabled_emits_metric_only(self) -> None: + cluster = redis.redis_clusters.get_binary("default") + keys_before = set(cluster.keys()) + + with ( + mock.patch.object(tasks, "metrics") as metrics, + mock.patch.object(repair_config_store, "delay") as delay, + ): + check_config_sentinels() + + # One gauge per store, both on the test cluster. + all_missing = mock.call( + "uptime.config_drift.sentinel_missing", + 128, + tags={"cluster": "default"}, + sample_rate=1.0, + ) + assert metrics.gauge.mock_calls == [all_missing, all_missing] + assert metrics.incr.mock_calls == [] + assert not delay.called + assert set(cluster.keys()) == keys_before + + @override_options({"uptime.config-drift.sentinel-repair-disabled": False}) + def test_sentinel_write_touches_no_config_hash(self) -> None: + cluster = redis.redis_clusters.get_binary("default") + + with self.tasks(): + check_config_sentinels() + + for key_prefix in "ab": + for partition in range(128): + assert cluster.type(get_sentinel_key(key_prefix, partition)) == b"string" + assert not cluster.exists(get_config_key(key_prefix, partition)) + assert not cluster.exists(f"{key_prefix}uptime:updates:{partition}") + + def _seed_lost_on_b(self, count: int = 1) -> tuple[list[UptimeSubscription], str]: + """ + Writes every sentinel, then creates ``count`` subscriptions never published to store B + and drops one of their partitions' sentinel there. Returns them and the lost sentinel key. + """ + with self.tasks(): + check_config_sentinels() + subscriptions = [] + for _ in range(count): + subscription = self.create_uptime_subscription( + subscription_id=_subscription_id(), region_slugs=["a1", "b1"] + ) + _publish(subscription, ["a1"]) + subscriptions.append(subscription) + assert subscription.subscription_id is not None + partition = get_partition_from_subscription_id(UUID(subscription.subscription_id)) + sentinel = get_sentinel_key("b", partition) + redis.redis_clusters.get_binary("default").delete(sentinel) + return subscriptions, sentinel + + @override_options({"uptime.config-drift.sentinel-repair-disabled": False}) + def test_missing_sentinel_repairs_only_that_store(self) -> None: + [subscription], _ = self._seed_lost_on_b() + + with mock.patch.object(repair_config_store, "delay") as delay: + check_config_sentinels() + delay.assert_called_once_with(cluster="default", key_prefix="b") + + with self.tasks(): + repair_config_store(cluster="default", key_prefix="b") + + self.assert_redis_config( + "b1", subscription, "upsert", UptimeSubscriptionRegion.RegionMode.ACTIVE + ) + + @override_options({"uptime.config-drift.sentinel-repair-disabled": False}) + @mock.patch.object(tasks, "CONFIG_REPAIR_MAX_TASKS", 1) + def test_unpublishable_configs_end_the_pass(self) -> None: + lost, sentinel = self._seed_lost_on_b(count=2) + first, second = sorted(lost, key=lambda subscription: subscription.subscription_id or "") + cluster = redis.redis_clusters.get_binary("default") + + # The republishes never land, so both stay missing on every run. + with mock.patch.object(update_remote_uptime_subscription, "delay") as delay: + repair_config_store(cluster="default", key_prefix="b") + assert not cluster.exists(sentinel) + repair_config_store(cluster="default", key_prefix="b") + + assert delay.call_args_list == [ + mock.call(uptime_subscription_id=first.id, region_slugs=["b1"]), + mock.call(uptime_subscription_id=second.id, region_slugs=["b1"]), + ] + assert cluster.exists(sentinel) + + with mock.patch.object(repair_config_store, "delay") as repair: + check_config_sentinels() + assert not repair.called + + @override_options({"uptime.config-drift.sentinel-repair-disabled": False}) + def test_leftover_cursor_does_not_skip_the_next_loss(self) -> None: + with self.tasks(): + check_config_sentinels() + # Left by an overlapping run after the sentinels came back, past every id. + redis.redis_clusters.get("default").set(get_repair_cursor_key("b"), "f" * 32, ex=60) + + # Its first check sees every sentinel present, which is where the cursor is cleared. + [subscription], _ = self._seed_lost_on_b() + with self.tasks(): + check_config_sentinels() + + self.assert_redis_config( + "b1", subscription, "upsert", UptimeSubscriptionRegion.RegionMode.ACTIVE + )