feat(crons): Add crons.use_clock_pulse_consumer option (#69531)

This will control producing clock pulse messages into a kafka topic
instead of dispatching to celery

Author: evanpurkhiser

src/sentry/monitors/clock_dispatch.py

@@ -6,6 +6,7 @@
 import sentry_sdk
 from django.conf import settings
 
+from sentry import options
 from sentry.monitors.tasks.check_missed import check_missing
 from sentry.monitors.tasks.check_timeout import check_timeout
 from sentry.utils import metrics, redis
@@ -41,8 +42,13 @@ def _dispatch_tick(ts: datetime):
     sentry.io, when we deploy we restart the celery beat worker and it will
     skip any tasks it missed)
     """
-    check_missing.delay(current_datetime=ts)
-    check_timeout.delay(current_datetime=ts)
+    if options.get("crons.use_clock_pulse_consumer"):
+        # TODO(epurkhiser): This should dispatch the pulse as a message on the
+        # monitors-clock-pulse topic
+        pass
+    else:
+        check_missing.delay(current_datetime=ts)
+        check_timeout.delay(current_datetime=ts)
 
 
 def try_monitor_clock_tick(ts: datetime, partition: int):

src/sentry/options/defaults.py

@@ -1820,6 +1820,14 @@
 # Killswitch for monitor check-ins
 register("crons.organization.disable-check-in", type=Sequence, default=[])
 
+# Controls the method of clock task dispatch. This is part of GH-58410 and will
+# enable dispatching via the clock pulse consumer instead of using celery tassk
+register(
+    "crons.use_clock_pulse_consumer",
+    default=False,
+    flags=FLAG_MODIFIABLE_BOOL | FLAG_AUTOMATOR_MODIFIABLE,
+)
+
 # Turns on and off the running for dynamic sampling collect_orgs.
 register("dynamic-sampling.tasks.collect_orgs", default=False, flags=FLAG_MODIFIABLE_BOOL)