PTFE-1196 reduce impact of missed webhooks (#491)

When a webhook is missed by the runner-manager whatever the reason may
be, instead of continuously being X runner late for a given group, we
continuously check the time it took for a given job to start and create
an extra runner if the value is too high.

While this may not completely solve the issue, it can reduce its impact
on users.

Author: tcarmet

.devcontainer/devcontainer.json

@@ -56,7 +56,7 @@
             "typeCheckingMode": "basic"
           },
           "editor": {
-            "defaultFormatter": "ms-python.python"
+            "defaultFormatter": "ms-python.black-formatter"
           }
         }
       }

runner_manager/jobs/workflow_job.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+from datetime import timedelta
 
 from githubkit.webhooks.models import (
     WorkflowJobCompleted,
@@ -9,8 +10,9 @@
 )
 from githubkit.webhooks.types import WorkflowJobEvent
 
+from runner_manager import Settings
 from runner_manager.clients.github import GitHub
-from runner_manager.dependencies import get_github
+from runner_manager.dependencies import get_github, get_settings
 from runner_manager.models.runner import Runner
 from runner_manager.models.runner_group import RunnerGroup
 
@@ -25,6 +27,11 @@ def log_workflow_job(webhook: WorkflowJobEvent) -> None:
     )
 
 
+def time_to_start(webhook: WorkflowJobInProgress | WorkflowJobCompleted) -> timedelta:
+    """From a given webhook, calculate the time it took to start the job"""
+    return webhook.workflow_job.started_at - webhook.workflow_job.created_at
+
+
 def completed(webhook: WorkflowJobCompleted) -> int:
     log_workflow_job(webhook)
     runner: Runner | None = Runner.find_from_webhook(webhook)
@@ -47,6 +54,7 @@ def completed(webhook: WorkflowJobCompleted) -> int:
 
 def in_progress(webhook: WorkflowJobInProgress) -> str | None:
     log_workflow_job(webhook)
+    settings: Settings = get_settings()
     name: str | None = webhook.workflow_job.runner_name
     runner_group: RunnerGroup | None = RunnerGroup.find_from_webhook(webhook)
     if not runner_group:
@@ -55,6 +63,17 @@ def in_progress(webhook: WorkflowJobInProgress) -> str | None:
     log.info(f"Updating runner {name} in group {runner_group.name}")
     runner: Runner = runner_group.update_runner(webhook=webhook)
     log.info(f"Runner {name} in group {runner_group.name} has been updated")
+    tts = time_to_start(webhook)
+    log.info(f"{runner} took {tts} to start")
+    # If the time to start is greater than settings.timeout_runner,
+    # create an extra runner.
+    # The main reason we perform this action is to ensure that
+    # in the case we have missed a webhook, we still have a runner
+    # available for the jobs that are requesting it.
+    if tts > settings.timeout_runner and runner_group.is_full is False:
+        log.info(f"Runner group {runner_group.name} needs a new runner")
+        github: GitHub = get_github()
+        runner_group.create_runner(github)
     return runner.pk
 
 

runner_manager/models/runner_group.py

@@ -219,6 +219,11 @@ def need_new_runner(self) -> bool:
         count = len(runners)
         return (not_active < self.min or self.queued > 0) and count < self.max
 
+    @property
+    def is_full(self) -> bool:
+        """Return True if the max number of runners has been reached."""
+        return len(self.get_runners()) >= self.max
+
     def create_github_group(self, github: GitHub) -> GitHubRunnerGroup:
         """Create a GitHub runner group."""
 

runner_manager/models/settings.py

@@ -31,7 +31,7 @@ class Settings(BaseSettings):
     api_key: Optional[SecretStr] = None
     log_level: Literal["INFO", "WARNING", "DEBUG", "ERROR"] = "INFO"
     runner_groups: List[BaseRunnerGroup] = []
-    timeout_runner: Optional[timedelta] = timedelta(minutes=15)
+    timeout_runner: timedelta = timedelta(minutes=15)
     time_to_live: Optional[timedelta] = timedelta(hours=12)
     healthcheck_interval: timedelta = timedelta(minutes=15)
     github_base_url: Optional[AnyHttpUrl] = Field(default="https://api.github.com")

tests/strategies.py

@@ -80,6 +80,7 @@ class Repo(Repository):
     runner_group_id=Int,
     labels=st.lists(Text, min_size=1, max_size=5),
     started_at=st.datetimes(),
+    created_at=st.datetimes(),
 )
 
 JobPropQueuedStrategy = st.builds(
@@ -127,7 +128,7 @@ class Repo(Repository):
     github_base_url=st.just("http://localhost:4010"),
     github_token=st.just("test"),
     time_to_live=st.integers(1, 60),
-    timeout_runner=st.integers(1, 10),
+    timeout_runner=st.integers(120, 600),
 )
 
 RedisStrategy = st.builds(

tests/unit/jobs/test_workflow_job.py

@@ -1,3 +1,4 @@
+from datetime import timedelta
 from time import sleep
 from uuid import uuid4
 
@@ -103,7 +104,6 @@ def test_workflow_job_completed(
 def test_workflow_job_in_progress(
     webhook: WorkflowJobInProgress, queue: Queue, settings: Settings, redis: Redis
 ):
-
     # flush all keys that start with settings.name in redis
 
     init_model(Runner, redis, settings)
@@ -197,3 +197,61 @@ def test_workflow_job_queued(
     ).first()
     assert runner.busy is False
     assert runner.status == "offline"
+
+
+@settings(max_examples=10)
+@given(
+    webhook=WorkflowJobInProgressStrategy,
+    queue=QueueStrategy,
+    settings=SettingsStrategy,
+    redis=RedisStrategy,
+)
+def test_workflow_job_time_to_start(
+    webhook: WorkflowJobInProgress, queue: Queue, settings: Settings, redis: Redis
+):
+    """
+    This test will ensure that an extra runner is created when the time to start
+    the given workflow was higher than settings.timeout_runners.
+    """
+    init_model(Runner, redis, settings)
+    init_model(RunnerGroup, redis, settings)
+    runner_group: RunnerGroup = RunnerGroup(
+        organization=webhook.organization.login,
+        name=webhook.workflow_job.runner_group_name,
+        id=webhook.workflow_job.runner_group_id,
+        labels=webhook.workflow_job.labels,
+        manager=settings.name,
+        backend={"name": "base"},
+        max=2,
+    )
+    runner_group.save()
+    Migrator().run()
+
+    runner: Runner = Runner(
+        id=webhook.workflow_job.runner_id,
+        name=webhook.workflow_job.runner_name,
+        busy=False,
+        status="online",
+        manager=settings.name,
+        runner_group_id=webhook.workflow_job.runner_group_id,
+        runner_group_name=webhook.workflow_job.runner_group_name,
+    )
+    runner.save()
+    Migrator().run()
+
+    assert len(runner_group.get_runners()) == 1
+    # ensure we have only one runner if the time to start is less than timeout_runners
+    webhook.workflow_job.started_at = webhook.workflow_job.created_at + (
+        settings.timeout_runner - timedelta(minutes=15)
+    )
+    queue.enqueue(workflow_job.in_progress, webhook)
+    assert len(runner_group.get_runners()) == 1
+    # ensure we have two runners if the time to start is greater than timeout_runners
+    webhook.workflow_job.started_at = webhook.workflow_job.created_at + (
+        settings.timeout_runner + timedelta(minutes=15)
+    )
+    queue.enqueue(workflow_job.in_progress, webhook)
+    assert len(runner_group.get_runners()) == 2
+    # ensure we remain with two runners given that the max for the runner group is 2
+    queue.enqueue(workflow_job.in_progress, webhook)
+    assert len(runner_group.get_runners()) == 2

tests/unit/models/test_runner_group.py

@@ -214,3 +214,14 @@ def test_find_github_group(runner_group: RunnerGroup, github: GitHub):
     assert exists is not None
     group = runner_group.save(github=github)
     assert exists.id == group.id
+
+
+def test_is_full(runner_group: RunnerGroup, github: GitHub):
+    runner_group.max = 2
+    runner_group.min = 1
+    runner_group.save()
+    assert runner_group.is_full is False
+    runner_group.create_runner(github)
+    assert runner_group.is_full is False
+    runner_group.create_runner(github)
+    assert runner_group.is_full is True