feat(taskworker) Add option to disable fetch_next behavior (#94994)

We have a few processing pools that are showing higher than desirable
contention on the child_task queue. Our theory is that having both the
main worker loop + fetch_next flow appending into the child_task queue
is causing contention resulting in lower throughput.

These changes introduce an option that allow us to disable fetch_next
logic at the pool level, this will allow us to confirm two theories:

1. That fetch_next is contributing to contention.
2. That we can get more throughput and use brokers more effectively as
GetTask and SetTaskStatus without fetch_next are faster than with
fetch_next.

I've reduced the maximum timeout for fetch_task to help reduce
processing latency when brokers are running on low throughput topics,
and cleaned up some now unused options.

Author: markstory

src/sentry/options/defaults.py

@@ -3186,32 +3186,24 @@
 
 # Taskbroker flags
 register(
-    "taskworker.try_compress.profile_metrics",
-    default=0.0,
-    type=Float,
-    flags=FLAG_AUTOMATOR_MODIFIABLE,
-)
-
-register(
-    "taskworker.try_compress.profile_metrics.rollout",
-    default=0.0,
-    type=Float,
+    "taskworker.route.overrides",
+    default={},
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
-
-# Taskbroker flags
 register(
     "taskworker.try_compress.profile_metrics.level",
     default=6,
     type=Int,
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
-
 register(
-    "taskworker.route.overrides",
-    default={},
+    "taskworker.fetch_next.disabled_pools",
+    default=[],
     flags=FLAG_AUTOMATOR_MODIFIABLE,
 )
+
+
+# Taskbroker rollout flags
 register(
     "taskworker.deletions.rollout",
     default={},

src/sentry/taskworker/worker.py

@@ -15,6 +15,7 @@
 from django.conf import settings
 from sentry_protos.taskbroker.v1.taskbroker_pb2 import FetchNextTask
 
+from sentry import options
 from sentry.taskworker.client.client import HostTemporarilyUnavailable, TaskworkerClient
 from sentry.taskworker.client.inflight_task_activation import InflightTaskActivation
 from sentry.taskworker.client.processing_result import ProcessingResult
@@ -177,6 +178,10 @@ def _add_task(self) -> bool:
                     tags={"processing_pool": self._processing_pool_name},
                 )
             except queue.Full:
+                metrics.incr(
+                    "taskworker.worker.child_tasks.put.full",
+                    tags={"processing_pool": self._processing_pool_name},
+                )
                 logger.warning(
                     "taskworker.add_task.child_task_queue_full",
                     extra={
@@ -204,9 +209,13 @@ def result_thread() -> None:
             iopool = ThreadPoolExecutor(max_workers=self._concurrency)
             with iopool as executor:
                 while not self._shutdown_event.is_set():
+                    fetch_next = self._processing_pool_name not in options.get(
+                        "taskworker.fetch_next.disabled_pools"
+                    )
+
                     try:
                         result = self._processed_tasks.get(timeout=1.0)
-                        executor.submit(self._send_result, result)
+                        executor.submit(self._send_result, result, fetch_next)
                     except queue.Empty:
                         metrics.incr(
                             "taskworker.worker.result_thread.queue_empty",
@@ -342,7 +351,7 @@ def fetch_task(self) -> InflightTaskActivation | None:
                 extra={"error": e, "processing_pool": self._processing_pool_name},
             )
 
-            self._gettask_backoff_seconds = min(self._gettask_backoff_seconds + 2, 10)
+            self._gettask_backoff_seconds = min(self._gettask_backoff_seconds + 1, 5)
             return None
 
         if not activation:
@@ -354,9 +363,7 @@ def fetch_task(self) -> InflightTaskActivation | None:
                 "taskworker.fetch_task.not_found",
                 extra={"processing_pool": self._processing_pool_name},
             )
-
-            # TODO cap backoff to 5 seconds instead?
-            self._gettask_backoff_seconds = min(self._gettask_backoff_seconds + 1, 10)
+            self._gettask_backoff_seconds = min(self._gettask_backoff_seconds + 1, 5)
             return None
 
         self._gettask_backoff_seconds = 0

src/sentry/taskworker/workerchild.py

@@ -171,16 +171,9 @@ def handle_alarm(signum: int, frame: FrameType | None) -> None:
                 )
                 break
 
-            child_tasks_get_start = time.monotonic()
             try:
-                # If the queue is empty, this could block for a second.
                 inflight = child_tasks.get(timeout=1.0)
             except queue.Empty:
-                metrics.distribution(
-                    "taskworker.worker.child_task_queue_empty.wait_duration",
-                    time.monotonic() - child_tasks_get_start,
-                    tags={"processing_pool": processing_pool_name},
-                )
                 metrics.incr(
                     "taskworker.worker.child_task_queue_empty",
                     tags={"processing_pool": processing_pool_name},

tests/sentry/processing/backpressure/test_checking.py

@@ -30,7 +30,6 @@
         "backpressure.checking.interval": 5,
         "backpressure.monitoring.enabled": True,
         "backpressure.status_ttl": 60,
-        "taskworker.try_compress.profile_metrics.rollout": 0,
     }
 )
 def test_backpressure_unhealthy_profiles():
@@ -54,7 +53,6 @@ def test_backpressure_unhealthy_profiles():
         "backpressure.checking.interval": 5,
         "backpressure.monitoring.enabled": False,
         "backpressure.status_ttl": 60,
-        "taskworker.try_compress.profile_metrics.rollout": 0,
     }
 )
 def test_bad_config():

tests/sentry/taskworker/test_worker.py

@@ -224,6 +224,43 @@ def update_task_response(*args, **kwargs):
             )
             assert mock_client.update_task.call_args.args[1] is None
 
+    @override_options({"taskworker.fetch_next.disabled_pools": ["testing"]})
+    def test_run_once_with_fetch_next_disabled(self) -> None:
+        # Cover the scenario where taskworker.fetch_next.disabled_pools is defined
+        max_runtime = 5
+        taskworker = TaskWorker(
+            rpc_host="127.0.0.1:50051",
+            num_brokers=1,
+            max_child_task_count=1,
+            process_type="fork",
+            processing_pool_name="testing",
+        )
+        with mock.patch.object(taskworker, "client") as mock_client:
+            mock_client.update_task.return_value = None
+            mock_client.get_task.return_value = SIMPLE_TASK
+            taskworker.start_result_thread()
+            taskworker.start_spawn_children_thread()
+
+            # Run until two tasks have been processed
+            start = time.time()
+            while True:
+                taskworker.run_once()
+                if mock_client.update_task.call_count >= 2:
+                    break
+                if time.time() - start > max_runtime:
+                    taskworker.shutdown()
+                    raise AssertionError("Timeout waiting for update_task to be called")
+
+            taskworker.shutdown()
+            assert mock_client.get_task.called
+            assert mock_client.update_task.call_count == 2
+            assert mock_client.update_task.call_args.args[0].host == "localhost:50051"
+            assert mock_client.update_task.call_args.args[0].task_id == SIMPLE_TASK.activation.id
+            assert (
+                mock_client.update_task.call_args.args[0].status == TASK_ACTIVATION_STATUS_COMPLETE
+            )
+            assert mock_client.update_task.call_args.args[1] is None
+
     def test_run_once_with_update_failure(self) -> None:
         # Cover the scenario where update_task fails a few times in a row
         # We should retain the result until RPC succeeds.