feat(taskworker): Make ingest errors taskworker compatible (#90340)

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

@@ -1406,6 +1406,11 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     "sentry.deletions.tasks.hybrid_cloud",
     "sentry.deletions.tasks.scheduled",
     "sentry.demo_mode.tasks",
+    "sentry.dynamic_sampling.tasks.boost_low_volume_projects",
+    "sentry.dynamic_sampling.tasks.boost_low_volume_transactions",
+    "sentry.dynamic_sampling.tasks.custom_rule_notifications",
+    "sentry.dynamic_sampling.tasks.recalibrate_orgs",
+    "sentry.dynamic_sampling.tasks.sliding_window_org",
     "sentry.hybridcloud.tasks.deliver_from_outbox",
     "sentry.hybridcloud.tasks.deliver_webhooks",
     "sentry.incidents.tasks",
@@ -1453,6 +1458,7 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     "sentry.tasks.auto_source_code_config",
     "sentry.tasks.autofix",
     "sentry.tasks.beacon",
+    "sentry.tasks.check_am2_compatibility",
     "sentry.tasks.check_new_issue_threshold_met",
     "sentry.tasks.clear_expired_resolutions",
     "sentry.tasks.clear_expired_rulesnoozes",
@@ -1465,14 +1471,15 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     "sentry.tasks.delete_seer_grouping_records",
     "sentry.tasks.digests",
     "sentry.tasks.email",
-    "sentry.tasks.relay",
     "sentry.tasks.embeddings_grouping.backfill_seer_grouping_records_for_project",
     "sentry.tasks.groupowner",
     "sentry.tasks.merge",
     "sentry.tasks.on_demand_metrics",
     "sentry.tasks.options",
     "sentry.tasks.ping",
+    "sentry.tasks.post_process",
     "sentry.tasks.process_buffer",
+    "sentry.tasks.relay",
     "sentry.tasks.release_registry",
     "sentry.tasks.repository",
     "sentry.tasks.reprocessing2",
@@ -1490,12 +1497,6 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     "sentry.uptime.rdap.tasks",
     "sentry.uptime.subscriptions.tasks",
     "sentry.workflow_engine.processors.delayed_workflow",
-    "sentry.dynamic_sampling.tasks.boost_low_volume_projects",
-    "sentry.dynamic_sampling.tasks.recalibrate_orgs",
-    "sentry.dynamic_sampling.tasks.custom_rule_notifications",
-    "sentry.dynamic_sampling.tasks.sliding_window_org",
-    "sentry.dynamic_sampling.tasks.boost_low_volume_transactions",
-    "sentry.tasks.check_am2_compatibility",
     # Used for tests
     "sentry.taskworker.tasks.examples",
 )
@@ -1508,6 +1509,13 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     },
 }
 
+TASKWORKER_ENABLE_HIGH_THROUGHPUT_NAMESPACES = False
+TASKWORKER_HIGH_THROUGHPUT_NAMESPACES = {
+    "ingest.profiling",
+    "ingest.transactions",
+    "ingest.errors",
+}
+
 # Sentry logs to two major places: stdout, and its internal project.
 # To disable logging to the internal project, add a logger whose only
 # handler is 'console' and disable propagating upwards.

src/sentry/options/defaults.py

@@ -3291,7 +3291,11 @@
     default={},
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
-
+register(
+    "taskworker.ingest.errors.rollout",
+    default={},
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)
 # Orgs for which compression should be disabled in the chunk upload endpoint.
 # This is intended to circumvent sporadic 503 errors reported by some customers.
 register("chunk-upload.no-compression", default=[], flags=FLAG_AUTOMATOR_MODIFIABLE)

src/sentry/tasks/base.py

@@ -75,14 +75,18 @@ def taskworker_override(
     def override(*args: P.args, **kwargs: P.kwargs) -> R:
         rollout_rate = 0
         option_flag = f"taskworker.{namespace}.rollout"
-        rollout_map = options.get(option_flag)
-        if rollout_map:
-            if task_name in rollout_map:
-                rollout_rate = rollout_map.get(task_name, 0)
-            elif "*" in rollout_map:
-                rollout_rate = rollout_map.get("*", 0)
-
-        random.seed(datetime.now().timestamp())
+        check_option = True
+        if namespace in settings.TASKWORKER_HIGH_THROUGHPUT_NAMESPACES:
+            check_option = settings.TASKWORKER_ENABLE_HIGH_THROUGHPUT_NAMESPACES
+
+        if check_option:
+            rollout_map = options.get(option_flag)
+            if rollout_map:
+                if task_name in rollout_map:
+                    rollout_rate = rollout_map.get(task_name, 0)
+                elif "*" in rollout_map:
+                    rollout_rate = rollout_map.get("*", 0)
+
         if rollout_rate > random.random():
             return taskworker_attr(*args, **kwargs)
 

src/sentry/tasks/post_process.py

@@ -25,6 +25,8 @@
 from sentry.signals import event_processed, issue_unignored
 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_errors_tasks
 from sentry.types.group import GroupSubStatus
 from sentry.utils import json, metrics
 from sentry.utils.cache import cache
@@ -483,6 +485,10 @@ def should_update_escalating_metrics(event: Event) -> bool:
     time_limit=120,
     soft_time_limit=110,
     silo_mode=SiloMode.REGION,
+    taskworker_config=TaskworkerConfig(
+        namespace=ingest_errors_tasks,
+        processing_deadline_duration=120,
+    ),
 )
 def post_process_group(
     is_new,

src/sentry/tasks/store.py

@@ -26,7 +26,7 @@
 from sentry.stacktraces.processing import process_stacktraces, should_process_for_stacktraces
 from sentry.tasks.base import instrumented_task
 from sentry.taskworker.config import TaskworkerConfig
-from sentry.taskworker.namespaces import issues_tasks
+from sentry.taskworker.namespaces import ingest_errors_tasks, issues_tasks
 from sentry.utils import metrics
 from sentry.utils.event_tracker import TransactionStageStatus, track_sampled_event
 from sentry.utils.safe import safe_execute
@@ -228,6 +228,10 @@ def _do_preprocess_event(
     time_limit=65,
     soft_time_limit=60,
     silo_mode=SiloMode.REGION,
+    taskworker_config=TaskworkerConfig(
+        namespace=ingest_errors_tasks,
+        processing_deadline_duration=65,
+    ),
 )
 def preprocess_event(
     cache_key: str,
@@ -438,6 +442,10 @@ def _continue_to_save_event() -> None:
     time_limit=65,
     soft_time_limit=60,
     silo_mode=SiloMode.REGION,
+    taskworker_config=TaskworkerConfig(
+        namespace=ingest_errors_tasks,
+        processing_deadline_duration=65,
+    ),
 )
 def process_event(
     cache_key: str,
@@ -632,6 +640,10 @@ def _do_save_event(
     time_limit=65,
     soft_time_limit=60,
     silo_mode=SiloMode.REGION,
+    taskworker_config=TaskworkerConfig(
+        namespace=ingest_errors_tasks,
+        processing_deadline_duration=65,
+    ),
 )
 def save_event(
     cache_key: str | None = None,

src/sentry/taskworker/namespaces.py

@@ -57,6 +57,8 @@
     app_feature="profiles",
 )
 
+ingest_errors_tasks = taskregistry.create_namespace("ingest.errors", app_feature="errors")
+
 issues_tasks = taskregistry.create_namespace("issues", app_feature="issueplatform")
 
 integrations_tasks = taskregistry.create_namespace("integrations", app_feature="integrations")