feat(taskworker): Make ingest profiling taskworker compatible (#90293)

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

@@ -1433,6 +1433,7 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     "sentry.monitors.tasks.clock_pulse",
     "sentry.monitors.tasks.detect_broken_monitor_envs",
     "sentry.notifications.utils.tasks",
+    "sentry.profiles.task",
     "sentry.release_health.tasks",
     "sentry.relocation.tasks.process",
     "sentry.relocation.tasks.transfer",
@@ -1471,13 +1472,12 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     "sentry.tasks.ping",
     "sentry.tasks.process_buffer",
     "sentry.tasks.release_registry",
-    "sentry.tasks.release_registry",
     "sentry.tasks.repository",
     "sentry.tasks.reprocessing2",
     "sentry.tasks.statistical_detectors",
-    "sentry.tasks.summaries.weekly_reports",
-    "sentry.tasks.summaries.daily_summary",
     "sentry.tasks.store",
+    "sentry.tasks.summaries.daily_summary",
+    "sentry.tasks.summaries.weekly_reports",
     "sentry.tasks.symbolication",
     "sentry.tasks.unmerge",
     "sentry.tasks.update_user_reports",

src/sentry/options/defaults.py

@@ -3366,6 +3366,11 @@
     default={},
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
+register(
+    "taskworker.ingest.profiling.rollout",
+    default={},
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)
 register(
     "taskworker.telemetry-experience.rollout",
     default={},

src/sentry/profiles/task.py

@@ -38,6 +38,9 @@
 from sentry.signals import first_profile_received
 from sentry.silo.base import SiloMode
 from sentry.tasks.base import instrumented_task
+from sentry.taskworker.config import TaskworkerConfig
+from sentry.taskworker.namespaces import ingest_profiling_tasks
+from sentry.taskworker.retry import Retry
 from sentry.utils import json, metrics
 from sentry.utils.arroyo_producer import SingletonProducer
 from sentry.utils.kafka_config import get_kafka_producer_cluster_options, get_topic_definition
@@ -88,6 +91,11 @@ def _get_profiles_producer_from_topic(topic: Topic) -> KafkaProducer:
     task_time_limit=60,
     task_acks_on_failure_or_timeout=False,
     silo_mode=SiloMode.REGION,
+    taskworker_config=TaskworkerConfig(
+        namespace=ingest_profiling_tasks,
+        processing_deadline_duration=60,
+        retry=Retry(times=2),
+    ),
 )
 def process_profile_task(
     profile: Profile | None = None,

src/sentry/taskworker/namespaces.py

@@ -34,6 +34,8 @@
 
 hybridcloud_control_tasks = taskregistry.create_namespace("hybridcloud.control")
 
+ingest_profiling_tasks = taskregistry.create_namespace("ingest.profiling")
+
 issues_tasks = taskregistry.create_namespace("issues")
 
 integrations_tasks = taskregistry.create_namespace("integrations")