feat(crons): Remove legacy celery based tasks (#70288)

Author: evanpurkhiser

src/sentry/conf/server.py

@@ -731,8 +731,6 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     "sentry.snuba.tasks",
     "sentry.replays.tasks",
     "sentry.monitors.tasks.clock_pulse",
-    "sentry.monitors.tasks.check_missed",
-    "sentry.monitors.tasks.check_timeout",
     "sentry.monitors.tasks.detect_broken_monitor_envs",
     "sentry.tasks.app_store_connect",
     "sentry.tasks.assemble",

src/sentry/monitors/clock_dispatch.py

@@ -3,17 +3,13 @@
 import logging
 from datetime import datetime, timedelta, timezone
 
-import sentry_sdk
 from arroyo import Topic as ArroyoTopic
 from arroyo.backends.kafka import KafkaPayload, KafkaProducer, build_kafka_configuration
 from django.conf import settings
 from sentry_kafka_schemas.codecs import Codec
 from sentry_kafka_schemas.schema_types.monitors_clock_tick_v1 import ClockTick
 
-from sentry import options
 from sentry.conf.types.kafka_definition import Topic, get_topic_codec
-from sentry.monitors.tasks.check_missed import check_missing
-from sentry.monitors.tasks.check_timeout import check_timeout
 from sentry.utils import metrics, redis
 from sentry.utils.arroyo_producer import SingletonProducer
 from sentry.utils.kafka_config import get_kafka_producer_cluster_options, get_topic_definition
@@ -63,19 +59,15 @@ def _dispatch_tick(ts: datetime):
     sentry.io, when we deploy we restart the celery beat worker and it will
     skip any tasks it missed)
     """
-    if options.get("crons.use_clock_pulse_consumer"):
-        if settings.SENTRY_EVENTSTREAM != "sentry.eventstream.kafka.KafkaEventStream":
-            # XXX(epurkhiser): Unclear what we want to do if we're not using kafka
-            return
+    if settings.SENTRY_EVENTSTREAM != "sentry.eventstream.kafka.KafkaEventStream":
+        # XXX(epurkhiser): Unclear what we want to do if we're not using kafka
+        return
 
-        message: ClockTick = {"ts": ts.timestamp()}
-        payload = KafkaPayload(None, CLOCK_TICK_CODEC.encode(message), [])
+    message: ClockTick = {"ts": ts.timestamp()}
+    payload = KafkaPayload(None, CLOCK_TICK_CODEC.encode(message), [])
 
-        topic = get_topic_definition(Topic.MONITORS_CLOCK_TICK)["real_topic_name"]
-        _clock_tick_producer.produce(ArroyoTopic(topic), payload)
-    else:
-        check_missing.delay(current_datetime=ts)
-        check_timeout.delay(current_datetime=ts)
+    topic = get_topic_definition(Topic.MONITORS_CLOCK_TICK)["real_topic_name"]
+    _clock_tick_producer.produce(ArroyoTopic(topic), payload)
 
 
 def try_monitor_clock_tick(ts: datetime, partition: int):
@@ -150,20 +142,14 @@ def try_monitor_clock_tick(ts: datetime, partition: int):
     # task run, backfill those ticks. This can happen when one partition has
     # slowed down too much and is missing a minutes worth of check-ins
     if last_ts is not None and slowest_part_ts > last_ts + 60:
-        if options.get("crons.use_clock_pulse_consumer"):
-            # We only want to do backfills when we're using the clock tick
-            # consumer, otherwise the celery tasks may process out of order
-            backfill_tick = datetime.fromtimestamp(last_ts + 60, tz=timezone.utc)
-            while backfill_tick < tick:
-                extra = {"reference_datetime": str(backfill_tick)}
-                logger.info("monitors.consumer.clock_tick_backfill", extra=extra)
-
-                _dispatch_tick(backfill_tick)
-                backfill_tick = backfill_tick + timedelta(minutes=1)
-        else:
-            with sentry_sdk.push_scope() as scope:
-                scope.set_extra("last_ts", last_ts)
-                scope.set_extra("slowest_part_ts", slowest_part_ts)
-                sentry_sdk.capture_message("Monitor task dispatch minute skipped")
+        # We only want to do backfills when we're using the clock tick
+        # consumer, otherwise the celery tasks may process out of order
+        backfill_tick = datetime.fromtimestamp(last_ts + 60, tz=timezone.utc)
+        while backfill_tick < tick:
+            extra = {"reference_datetime": str(backfill_tick)}
+            logger.info("monitors.consumer.clock_tick_backfill", extra=extra)
+
+            _dispatch_tick(backfill_tick)
+            backfill_tick = backfill_tick + timedelta(minutes=1)
 
     _dispatch_tick(tick)

src/sentry/monitors/tasks/check_missed.py

@@ -74,10 +74,6 @@
     "sentry.profiles.task.process_profile": 0.01,
     "sentry.tasks.derive_code_mappings.process_organizations": settings.SAMPLED_DEFAULT_RATE,
     "sentry.tasks.derive_code_mappings.derive_code_mappings": settings.SAMPLED_DEFAULT_RATE,
-    "sentry.monitors.tasks.check_missing": 1.0,
-    "sentry.monitors.tasks.mark_environment_missing": 0.05,
-    "sentry.monitors.tasks.check_timeout": 1.0,
-    "sentry.monitors.tasks.mark_checkin_timeout": 0.05,
     "sentry.monitors.tasks.clock_pulse": 1.0,
     "sentry.tasks.auto_enable_codecov": settings.SAMPLED_DEFAULT_RATE,
     "sentry.dynamic_sampling.tasks.boost_low_volume_projects": 1.0,

src/sentry/monitors/tasks/check_timeout.py

@@ -26,7 +26,6 @@
     ScheduleType,
 )
 from sentry.testutils.cases import TestCase
-from sentry.testutils.helpers.options import override_options
 
 partition = Partition(Topic("test"), 0)
 
@@ -61,7 +60,6 @@ def test_simple(mock_dispatch_check_timeout, mock_dispatch_check_missing):
 
 class MonitorsClockTickEndToEndTest(TestCase):
     @override_settings(SENTRY_EVENTSTREAM="sentry.eventstream.kafka.KafkaEventStream")
-    @override_options({"crons.use_clock_pulse_consumer": True})
     def test_end_to_end(self):
         ts = timezone.now().replace(second=0, microsecond=0)