feat: parallel deletion of runners (#588)

* feat: parallel deletion of runners

* chore: minor refactor for imports

* feat: add delete vms and extract metrics interface

* feat: add delete runners interface

* feat: delete runner controller implementation

* feat: delete runner implementation for GitHub platform

* feat: delete runner implementation on jobmanager

* feat: parallel vm deletion and metrics extraction

* feat: cloud manager + cloud delete vms implementation

* chore: reorder helper func

* test: runner pull metrics

* feat: replace singular delete with parallel deletes

* chore: ignore docstring errors in abc

* chore: lint fixes

* test: redo tests

* test: fix openstack fixture

* chore: log extracted metric s

* debug

* chore: debug log

* chore: try logging with file

* general exception catch w/ traceback

* fix: construct connection on subprocess call

* chore: minor lint fixes

* chore: remove comments

* chore: revert debug workflow

* chore: tidy up factory comments

* chore: remove reusing connection

* fix: pyproject merge conflict fix

* fix: adapt changes to SelfHostedRunnerLabels

* fix: pyproject toml conflicting test ignores

* chore: remove todo comments

* feat: use thread pool for concurrency

* feat: limit max workers

* feat: concurrent github runner delete request using threadpool

* feat: concurrent metrics fetching using multithreading

* chore: rename DeleteVMError

* chore: update unused argument comment

* chore: rename mock to fake

* fix: lint issues w docstring

* debug

* fix: instantiate vars before assignment

* chore: delete unused test bin

* chore: pass down delete timeout

* chore: update keypair delete to fire and forget

* chore: undo debug

* fix: lint rules

* chore: fix meta docstring

* chore: remove unused platform argument after multiplexer ejection

* fix: result returns none

* fix: only return truely deleted openstack IDs

Author: yanksyoon

github-runner-manager/pyproject.toml

@@ -79,9 +79,15 @@ select = ["E", "W", "F", "C", "N", "R", "D", "H"]
 # Ignore W503,E203 because using black creates errors with this
 # Ignore D107 Missing docstring in __init__
 ignore = ["W503", "D107", "E203"]
-# D100, D101, D102, D103, D104: Ignore docstring style issues in tests
-# DCO020, DCO030, DCO050: Ignore docstring argument,returns,raises sections in tests
-per-file-ignores = ["tests/*:D100,D101,D102,D103,D104,D205,D212, DCO020, DCO030, DCO050"]
+per-file-ignores = [
+    # Ignore factory methods attributes docstring
+    "tests/unit/factories/*:DCO060",
+    # Ignore no return values (DCO031) in docstring for abstract methods
+    "src/github_runner_manager/manager/cloud_runner_manager.py:DCO031",
+    # DCO020, DCO030, DCO050, DCO060: Ignore docstring argument, returns, raises, attribute
+    # sections in tests
+    "tests/*:D100,D101,D102,D103,D104,D205,D212,DCO020,DCO030,DCO050,DCO060",
+]
 docstring-convention = "google"
 # Check for properly formatted copyright header in each file
 copyright-check = "True"

github-runner-manager/src/github_runner_manager/github_client.py

@@ -23,11 +23,7 @@
 from requests import RequestException
 from typing_extensions import assert_never
 
-from github_runner_manager.configuration.github import (
-    GitHubOrg,
-    GitHubPath,
-    GitHubRepo,
-)
+from github_runner_manager.configuration.github import GitHubOrg, GitHubPath, GitHubRepo
 from github_runner_manager.manager.models import InstanceID
 from github_runner_manager.platform.platform_provider import (
     DeleteRunnerBusyError,

github-runner-manager/src/github_runner_manager/manager/cloud_runner_manager.py

@@ -35,20 +35,6 @@ class HealthState(Enum):
     UNHEALTHY = auto()
     UNKNOWN = auto()
 
-    @staticmethod
-    def from_value(health: bool | None) -> "HealthState":
-        """Create from a health value.
-
-        Args:
-            health: The health value as boolean or None.
-
-        Returns:
-            The health state.
-        """
-        if health is None:
-            return HealthState.UNKNOWN
-        return HealthState.HEALTHY if health else HealthState.UNHEALTHY
-
 
 class CloudRunnerState(str, Enum):
     """Represent state of the instance hosting the runner.
@@ -275,11 +261,25 @@ def get_runners(self) -> Sequence[CloudRunnerInstance]:
         """Get cloud self-hosted runners."""
 
     @abc.abstractmethod
-    def delete_runner(self, instance_id: InstanceID) -> RunnerMetrics | None:
-        """Delete self-hosted runner.
+    def delete_vms(self, instance_ids: Sequence[InstanceID]) -> list[InstanceID]:
+        """Delete cloud VM instances.
+
+        Args:
+            instance_ids: The ID of the VMs to request deletion.
+
+        Returns:
+            The deleted instance IDs.
+        """
+
+    @abc.abstractmethod
+    def extract_metrics(self, instance_ids: Sequence[InstanceID]) -> list[RunnerMetrics]:
+        """Extract metrics from cloud VMs.
 
         Args:
-            instance_id: The instance id of the runner to delete.
+            instance_ids: The VM instance IDs to fetch the metrics from.
+
+        Returns:
+            The fetched runner metrics.
         """
 
     @abc.abstractmethod

github-runner-manager/src/github_runner_manager/manager/models.py

@@ -32,8 +32,8 @@ class InstanceID:
     """
 
     prefix: str
-    reactive: bool | None
     suffix: str
+    reactive: bool | None = None
 
     @property
     def name(self) -> str:

github-runner-manager/src/github_runner_manager/manager/runner_manager.py

@@ -23,11 +23,9 @@
 from github_runner_manager.metrics import events as metric_events
 from github_runner_manager.metrics import github as github_metrics
 from github_runner_manager.metrics import runner as runner_metrics
-from github_runner_manager.metrics.reconcile import CLEANED_RUNNERS_TOTAL
 from github_runner_manager.metrics.runner import RunnerMetrics
 from github_runner_manager.openstack_cloud.constants import CREATE_SERVER_TIMEOUT
 from github_runner_manager.platform.platform_provider import (
-    DeleteRunnerBusyError,
     PlatformApiError,
     PlatformProvider,
     PlatformRunnerHealth,
@@ -82,27 +80,33 @@ class RunnerInstance:
     platform_state: PlatformRunnerState | None
     cloud_state: CloudRunnerState
 
-    def __init__(
-        self,
+    @classmethod
+    def from_cloud_and_platform_health(
+        cls,
         cloud_instance: CloudRunnerInstance,
         platform_health_state: PlatformRunnerHealth | None,
-    ):
+    ) -> "RunnerInstance":
         """Construct an instance.
 
         Args:
             cloud_instance: Information on the cloud instance.
             platform_health_state: Health state in the platform provider.
+
+        Returns:
+            The RunnerInstance instantiated from cloud instance and platform state.
         """
-        self.name = cloud_instance.name
-        self.instance_id = cloud_instance.instance_id
-        self.metadata = cloud_instance.metadata
-        self.health = cloud_instance.health
-        self.platform_state = (
-            PlatformRunnerState.from_platform_health(platform_health_state)
-            if platform_health_state is not None
-            else None
+        return cls(
+            name=cloud_instance.name,
+            instance_id=cloud_instance.instance_id,
+            metadata=cloud_instance.metadata,
+            health=cloud_instance.health,
+            platform_state=(
+                PlatformRunnerState.from_platform_health(platform_health_state)
+                if platform_health_state is not None
+                else None
+            ),
+            cloud_state=cloud_instance.state,
         )
-        self.cloud_state = cloud_instance.state
 
 
 class RunnerManager:
@@ -183,7 +187,7 @@ def get_runners(self) -> tuple[RunnerInstance, ...]:
         health_runners_map = {runner.identity.instance_id: runner for runner in runners_health}
         for cloud_runner in cloud_runners:
             if cloud_runner.instance_id not in health_runners_map:
-                runner_instance = RunnerInstance(cloud_runner, None)
+                runner_instance = RunnerInstance.from_cloud_and_platform_health(cloud_runner, None)
                 runner_instance.health = HealthState.UNKNOWN
                 runner_instances.append(runner_instance)
                 continue
@@ -194,7 +198,9 @@ def get_runners(self) -> tuple[RunnerInstance, ...]:
                 cloud_runner.health = HealthState.HEALTHY
             else:
                 cloud_runner.health = HealthState.UNHEALTHY
-            runner_instance = RunnerInstance(cloud_runner, health_runner)
+            runner_instance = RunnerInstance.from_cloud_and_platform_health(
+                cloud_runner, health_runner
+            )
             runner_instances.append(runner_instance)
         return cast(tuple[RunnerInstance], tuple(runner_instances))
 
@@ -268,7 +274,9 @@ def _cleanup_resources(
         )
         cloud_runners = self._cloud.get_runners()
         logger.info("cleanup cloud_runners %s", cloud_runners)
-        runners_health_response = self._platform.get_runners_health(cloud_runners)
+        runners_health_response = self._platform.get_runners_health(
+            requested_runners=cloud_runners
+        )
         logger.info("cleanup health_response %s", runners_health_response)
 
         # Clean dangling resources in the cloud
@@ -295,7 +303,7 @@ def _cleanup_resources(
             )
         )
 
-        if maximum_runners_to_delete:
+        if maximum_runners_to_delete is not None:
             cloud_runners_to_delete.sort(
                 key=partial(_runner_deletion_sort_key, health_runners_map)
             )
@@ -313,58 +321,70 @@ def _delete_cloud_runners(
         runners_health: Sequence[PlatformRunnerHealth],
         delete_busy_runners: bool = False,
     ) -> Iterable[runner_metrics.RunnerMetrics]:
-        """Delete runners in the platform ant the cloud.
+        """Delete runners in the platform and the cloud.
 
         If delete_busy_runners is False, when the platform provider fails in deleting the
         runner because it can be busy, will mean that that runner should not be deleted.
+
+        Runners without health information should not be deleted.
         """
-        extracted_runner_metrics = []
-        health_runners_map = {health.identity.instance_id: health for health in runners_health}
-        for cloud_runner in cloud_runners:
-            logging.info("Trying to delete cloud_runner %s", cloud_runner)
-            runner_health = health_runners_map.get(cloud_runner.instance_id)
-            if runner_health and runner_health.runner_in_platform:
-                try:
-                    self._platform.delete_runner(runner_health.identity)
-                except DeleteRunnerBusyError:
-                    if not delete_busy_runners:
-                        logger.warning(
-                            "Skipping deletion as the runner is busy. %s", cloud_runner.instance_id
-                        )
-                        continue
-                    logger.info("Deleting busy runner: %s", cloud_runner.instance_id)
-                except PlatformApiError as exc:
-                    if not delete_busy_runners:
-                        logger.warning(
-                            "Failed to delete platform runner %s. %s. Skipping.",
-                            cloud_runner.instance_id,
-                            exc,
-                        )
-                        continue
-                    logger.warning(
-                        "Deleting runner: %s after platform failure %s.",
-                        cloud_runner.instance_id,
-                        exc,
-                    )
+        if not cloud_runners:
+            return []
 
-            logging.info("Delete runner in cloud: %s", cloud_runner.instance_id)
-            runner_metric = self._cloud.delete_runner(cloud_runner.instance_id)
-            CLEANED_RUNNERS_TOTAL.labels(self.manager_name).inc(1)
-            if not runner_metric:
-                logger.error("No metrics returned after deleting %s", cloud_runner.instance_id)
-            else:
-                extracted_runner_metrics.append(runner_metric)
-        return extracted_runner_metrics
+        runner_identity_map = {
+            health_info.identity.instance_id: health_info.identity
+            for health_info in runners_health
+        }
+        platform_runner_ids_to_delete = [
+            # The runner_id cannot be None due to the if condition. the type system
+            # isn't able to catch that.
+            cast(str, runner_identity_map[runner.instance_id].metadata.runner_id)
+            for runner in cloud_runners
+            if runner.instance_id in runner_identity_map
+            and runner_identity_map[runner.instance_id].metadata.runner_id
+        ]
+        logger.info("Deleting runners from platform: %s", platform_runner_ids_to_delete)
+        deleted_runner_ids = self._platform.delete_runners(
+            runner_ids=platform_runner_ids_to_delete
+        )
+        logger.info(
+            "Deleted runners from platform: %s (diff: %s)",
+            deleted_runner_ids,
+            set(platform_runner_ids_to_delete) - set(deleted_runner_ids),
+        )
+
+        logger.info("Cloud runners: %s", cloud_runners)
+        cloud_vm_ids_to_delete = [
+            runner.instance_id
+            for runner in cloud_runners
+            # We can delete all VMs if delete_busy_runners is True
+            if delete_busy_runners
+            # We can delete the VM if no runner is associated with it
+            or not runner.metadata.runner_id
+            # We can delete the VM if it has been deleted from the Platform provider.
+            or runner.metadata.runner_id in deleted_runner_ids
+        ]
+        logger.info("Extracting metrics from cloud VMs: %s", cloud_vm_ids_to_delete)
+        extracted_metrics = self._cloud.extract_metrics(instance_ids=cloud_vm_ids_to_delete)
+        logger.info("Extracted metrics from cloud VMs: %s", extracted_metrics)
+        logger.info("Deleting VMs %s", cloud_vm_ids_to_delete)
+        deleted_vm_ids = self._cloud.delete_vms(instance_ids=cloud_vm_ids_to_delete)
+        logger.info(
+            "Deleted VMs: %s, (diff: %s)",
+            deleted_vm_ids,
+            set(cloud_vm_ids_to_delete) - set(deleted_vm_ids),
+        )
+        return tuple(extracted_metrics)
 
     def _clean_platform_runners(self, runners: list[RunnerIdentity]) -> None:
         """Clean the specified runners in the platform."""
-        for runner in runners:
-            try:
-                self._platform.delete_runner(runner)
-            except DeleteRunnerBusyError:
-                logger.warning("Tried to delete busy runner in cleanup %s", runner)
-            except PlatformApiError:
-                logger.warning("Failed to delete platform runner %s", runner)
+        if not runners:
+            return
+
+        runner_ids_to_delete = [
+            runner.metadata.runner_id for runner in runners if runner.metadata.runner_id
+        ]
+        self._platform.delete_runners(runner_ids=runner_ids_to_delete)
 
     @staticmethod
     def _spawn_runners(
@@ -525,7 +545,7 @@ def _create_runner(args: _CreateRunnerArgs) -> InstanceID:
             )
         except RunnerError:
             logger.warning("Deleting runner %s from platform after creation failed", instance_id)
-            args.platform_provider.delete_runner(runner_info.identity)
+            args.platform_provider.delete_runners(runner_ids=[args.metadata.runner_id])
             raise
         return instance_id
 

github-runner-manager/src/github_runner_manager/manager/runner_scaler.py

@@ -94,7 +94,7 @@ class _ReconcileMetricData:
     start_timestamp: float
     end_timestamp: float
     metric_stats: IssuedMetricEventsStats
-    runner_list: tuple[RunnerInstance]
+    runner_list: tuple[RunnerInstance, ...]
     flavor: str
     expected_runner_quantity: int
 
@@ -375,7 +375,7 @@ def _reconcile_non_reactive(self, expected_quantity: int) -> _ReconcileResult:
         return _ReconcileResult(runner_diff=runner_diff, metric_stats=metric_stats)
 
     @staticmethod
-    def _log_runners(runner_list: tuple[RunnerInstance]) -> None:
+    def _log_runners(runner_list: tuple[RunnerInstance, ...]) -> None:
         """Log information about the runners found.
 
         Args:
@@ -446,7 +446,6 @@ def _issue_reconciliation_metric(
     IDLE_RUNNERS_COUNT.labels(manager_name).set(len(idle_runners))
 
     try:
-
         metric_events.issue_event(
             metric_events.Reconciliation(
                 timestamp=time.time(),