[Swarming] Add backpressure mechanism to SwarmingService

[IvanBM18]

Overview

This PR implements a backpressure mechanism in SwarmingService to prevent overloading the Swarming pool with too many tasks.

Changes

• Added MAX_PENDING_TASKS = 25 to limit the number of pending tasks.
  • This threshold was chosen considering a pool of at most 25 bots and that a single scheduler takes ~3 min to schedule all of its tasks. And since we have 1000 schedulers in prod.. the best choice is to instead look for a stable backlog.
  • 25 ensures that the 25 bots always have a buffer of work without building up a massive backlog
• Updated create_utask_main_jobs to check the queue size before pushing each task using Swarming's CountTasks RPC.
• Implemented a Fail Closed strategy: if the CountTasks rpc call fails, we assume the queue is full and stop scheduling further tasks in the batch.
• Added unit tests to verify backpressure and failure handling.

Tests performed

In this image we can appreciate that the utask main schedulers picks tasks from both the utask_main queue and the new utask_main-swarming queue

Heres another logs showing us that non swarming tasks are still being scheduled:

Here are some logs of the scheduler pushing tasks into swarming:

[letitz]

One substantive comment about the queue size limit being per-OS, otherwise a bunch of small comments.

[letitz]

LGTM % a couple remaining comments to avoid blocking on me further.

[IvanBM18]

Note for reviewer:
Since the swarming queue is not a pub/sub queue i extracted some of the logic from the data class to the public to reuse it here.

The other alternative to this is create a PubSubTaskQueue instance for the swarming queue(im not against this!) but if i do so i would also like to rename the PubSubTaskQueue module, since it would be confusing to have a object instance that its not really pub/sub queue.

[letitz]

That makes sense, and it's good enough for me to land this PR. I think ideally you would have something like:

# task_queue.py

class TaskQueueLimiter:
  default_limit: int
  feature_flag: FeatureFlags

  def max_pending_tasks(self) -> int:
    ...


class PubSubTaskQueue:
  name: str
  limiter: TaskQueueLimiter

Since the module name here is still awkward. I don't think you should actually do it though, this is good enough and we have other things to do :)

[letitz]

Thanks, we're nearly there!

[letitz]

That makes sense, and it's good enough for me to land this PR. I think ideally you would have something like:

# task_queue.py

class TaskQueueLimiter:
  default_limit: int
  feature_flag: FeatureFlags

  def max_pending_tasks(self) -> int:
    ...


class PubSubTaskQueue:
  name: str
  limiter: TaskQueueLimiter

Since the module name here is still awkward. I don't think you should actually do it though, this is good enough and we have other things to do :)

[letitz]

LGTM % tiny fixes to SwarmingApi.count_tasks().

[javanlacerda]

lgtm!