feat(taskworker): Make crons taskworker compatible (#88727)

This work is required to migrate tasks from celery to the new taskbroker
system. The sentry option will be used to control the rollout of these
tasks. The full migration plan is describe in this
[document](https://www.notion.so/sentry/Rollout-Planning-1bd8b10e4b5d80aeaaa7dba0efca83bc).

Author: enochtangg

src/sentry/conf/server.py

@@ -1401,6 +1401,8 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     "sentry.deletions.tasks.hybrid_cloud",
     "sentry.deletions.tasks.scheduled",
     "sentry.incidents.tasks",
+    "sentry.monitors.tasks.clock_pulse",
+    "sentry.monitors.tasks.detect_broken_monitor_envs",
     "sentry.snuba.tasks",
     "sentry.tasks.auth.auth",
     "sentry.tasks.auth.check_auth",

src/sentry/monitors/tasks/clock_pulse.py

@@ -17,6 +17,8 @@
 from sentry.monitors.clock_dispatch import try_monitor_clock_tick
 from sentry.silo.base import SiloMode
 from sentry.tasks.base import instrumented_task
+from sentry.taskworker.config import TaskworkerConfig
+from sentry.taskworker.namespaces import crons_tasks
 from sentry.utils.arroyo_producer import SingletonProducer
 from sentry.utils.kafka_config import (
     get_kafka_admin_cluster_options,
@@ -54,7 +56,11 @@ def _get_partitions() -> Mapping[int, PartitionMetadata]:
     return topic_metadata.partitions
 
 
-@instrumented_task(name="sentry.monitors.tasks.clock_pulse", silo_mode=SiloMode.REGION)
+@instrumented_task(
+    name="sentry.monitors.tasks.clock_pulse",
+    silo_mode=SiloMode.REGION,
+    taskworker_config=TaskworkerConfig(namespace=crons_tasks),
+)
 def clock_pulse(current_datetime=None):
     """
     This task is run once a minute when to produce 'clock pulses' into the

src/sentry/monitors/tasks/detect_broken_monitor_envs.py

@@ -26,6 +26,8 @@
 from sentry.notifications.services import notifications_service
 from sentry.notifications.types import NotificationSettingEnum
 from sentry.tasks.base import instrumented_task
+from sentry.taskworker.config import TaskworkerConfig
+from sentry.taskworker.namespaces import crons_tasks
 from sentry.types.actor import Actor
 from sentry.utils.email import MessageBuilder
 from sentry.utils.email.manager import get_email_addresses
@@ -162,6 +164,7 @@ def build_open_incidents_queryset():
     time_limit=15 * 60,
     soft_time_limit=10 * 60,
     record_timing=True,
+    taskworker_config=TaskworkerConfig(namespace=crons_tasks, processing_deadline_duration=15 * 60),
 )
 def detect_broken_monitor_envs():
     open_incidents_qs = build_open_incidents_queryset()
@@ -179,6 +182,7 @@ def detect_broken_monitor_envs():
     time_limit=15 * 60,
     soft_time_limit=10 * 60,
     record_timing=True,
+    taskworker_config=TaskworkerConfig(namespace=crons_tasks, processing_deadline_duration=15 * 60),
 )
 def detect_broken_monitor_envs_for_org(org_id: int):
     current_time = django_timezone.now()

src/sentry/options/defaults.py

@@ -3235,3 +3235,8 @@
     default={},
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
+register(
+    "taskworker.crons.rollout",
+    default={},
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)

src/sentry/taskworker/namespaces.py

@@ -7,6 +7,8 @@
 
 auth_control_tasks = taskregistry.create_namespace("auth.control")
 
+crons_tasks = taskregistry.create_namespace("crons")
+
 deletion_tasks = taskregistry.create_namespace(
     "deletions",
     processing_deadline_duration=60 * 3,