diff --git a/gpustack_runtime/cmds/deployer.py b/gpustack_runtime/cmds/deployer.py index e473e5e..d626808 100644 --- a/gpustack_runtime/cmds/deployer.py +++ b/gpustack_runtime/cmds/deployer.py @@ -362,6 +362,7 @@ class DeleteWorkloadSubCommand(SubCommand): namespace: str name: str + grace_period_seconds: int | None @staticmethod def register(parser: _SubParsersAction): @@ -376,6 +377,13 @@ def register(parser: _SubParsersAction): help="Namespace of the workload", ) + delete_parser.add_argument( + "--grace-period-seconds", + type=int, + help="Duration in seconds the workload needs to terminate gracefully, " + "which overrides the one declared by the workload plan", + ) + delete_parser.add_argument( "name", type=str, @@ -387,6 +395,7 @@ def register(parser: _SubParsersAction): def __init__(self, args: Namespace): self.namespace = args.namespace self.name = args.name + self.grace_period_seconds = args.grace_period_seconds if not self.name: msg = "The name argument is required." @@ -396,6 +405,7 @@ def run(self): st = delete_workload( name=self.name, namespace=self.namespace, + grace_period_seconds=self.grace_period_seconds, ) if st: print(f"Deleted workload '{self.name}'.") @@ -428,11 +438,21 @@ def register(parser: _SubParsersAction): help="Filter workloads by labels (key=value pairs separated by commas)", ) + delete_parser.add_argument( + "--grace-period-seconds", + type=int, + help="Duration in seconds the workloads need to terminate gracefully, " + "which overrides the one declared by the workload plans. " + "The workloads are deleted one after another, " + "so the whole deletion may take this duration per workload", + ) + delete_parser.set_defaults(func=DeleteWorkloadsSubCommand) def __init__(self, args: Namespace): self.namespace = args.namespace self.labels = args.labels + self.grace_period_seconds = args.grace_period_seconds def run(self): sts: list[WorkloadStatus] = list_workloads( @@ -443,6 +463,7 @@ def run(self): delete_workload( name=st.name, namespace=st.namespace, + grace_period_seconds=self.grace_period_seconds, ) print(f"Deleted workload '{st.name}'.") if not sts: diff --git a/gpustack_runtime/deployer/__init__.py b/gpustack_runtime/deployer/__init__.py index 7a1d39a..ed12fd8 100644 --- a/gpustack_runtime/deployer/__init__.py +++ b/gpustack_runtime/deployer/__init__.py @@ -150,6 +150,7 @@ def get_workload( def delete_workload( name: WorkloadName, namespace: WorkloadNamespace | None = None, + grace_period_seconds: int | None = None, ) -> WorkloadStatus | None: """ Delete the given workload. @@ -159,11 +160,16 @@ def delete_workload( The name of the workload to delete. namespace: The namespace of the workload. + grace_period_seconds: + Duration in seconds the workload needs to terminate gracefully, + which overrides the one declared by the workload plan. Return: The status if found, None otherwise. Raises: + ValueError: + If the grace period is negative. UnsupportedError: If no deployer supports the given workload. OperationError: @@ -174,7 +180,11 @@ def delete_workload( if not dep.is_supported(): continue - return dep.delete(name=name, namespace=namespace) + return dep.delete( + name=name, + namespace=namespace, + grace_period_seconds=grace_period_seconds, + ) raise UnsupportedError(_NO_AVAILABLE_DEPLOYER_MSG) diff --git a/gpustack_runtime/deployer/__types__.py b/gpustack_runtime/deployer/__types__.py index f02bf5e..53cb6ea 100644 --- a/gpustack_runtime/deployer/__types__.py +++ b/gpustack_runtime/deployer/__types__.py @@ -832,6 +832,12 @@ class WorkloadSecurity: Name for a workload. """ +DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS = 30 +""" +Default duration in seconds a workload needs to terminate gracefully, +which aligns with the Kubernetes API server default. +""" + @dataclass_json @dataclass @@ -862,6 +868,8 @@ class WorkloadPlan(WorkloadSecurity): The group ID to own the filesystem of the workload. sysctls (dict[str, str] | None): Sysctls to set for the workload. + termination_grace_period_seconds (int): + Duration in seconds the containers of the workload need to terminate gracefully. containers (list[Container] | None): Containers in the workload. It must contain at least one "RUN" profile container. @@ -909,6 +917,12 @@ class WorkloadPlan(WorkloadSecurity): Containers in the workload. It must contain at least one "RUN" profile container. """ + termination_grace_period_seconds: int = DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS + """ + Duration in seconds the containers of the workload need to terminate gracefully. + Containers are signaled to stop, and killed once the duration elapses. + Zero means killing immediately. + """ ## ## The below are internal use only fields. @@ -957,6 +971,15 @@ def validate_and_default(self): msg = 'Workload must contain at least one "RUN" profile container.' raise ValueError(msg) + # Default and validate termination grace period. + if self.termination_grace_period_seconds is None: + self.termination_grace_period_seconds = ( + DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS + ) + elif self.termination_grace_period_seconds < 0: + msg = "Workload termination grace period must not be negative." + raise ValueError(msg) + # Validate workload labels, including label names and values. for ln, lv in self.labels.items(): for s in ln.split("/"): @@ -2109,6 +2132,8 @@ def delete( self, name: WorkloadName, namespace: WorkloadNamespace | None = None, + *, + grace_period_seconds: int | None = None, async_mode: bool | None = None, ) -> WorkloadStatus | None: """ @@ -2119,6 +2144,9 @@ def delete( The name of the workload. namespace: The namespace of the workload. + grace_period_seconds: + Duration in seconds the workload needs to terminate gracefully, + which overrides the one declared by the workload plan. async_mode: Whether to execute in a separate thread. @@ -2128,16 +2156,23 @@ def delete( Raises: UnsupportedError: If the deployer is not supported in the current environment. + ValueError: + If the grace period is negative. OperationError: If the workload fails to delete. """ + if grace_period_seconds is not None and grace_period_seconds < 0: + msg = "Workload termination grace period must not be negative." + raise ValueError(msg) + if async_mode: try: future = self.pool.submit( self._delete, name, namespace, + grace_period_seconds, ) return future.result() except OperationError: @@ -2146,13 +2181,14 @@ def delete( msg = "Asynchronous workload delete failed." raise OperationError(msg) from e else: - return self._delete(name, namespace) + return self._delete(name, namespace, grace_period_seconds) @abstractmethod def _delete( self, name: WorkloadName, namespace: WorkloadNamespace | None = None, + grace_period_seconds: int | None = None, ) -> WorkloadStatus | None: """ Delete a workload. @@ -2162,6 +2198,9 @@ def _delete( The name of the workload. namespace: The namespace of the workload. + grace_period_seconds: + Duration in seconds the workload needs to terminate gracefully, + which overrides the one declared by the workload plan. Return: The status if found, None otherwise. diff --git a/gpustack_runtime/deployer/docker.py b/gpustack_runtime/deployer/docker.py index 7c85369..e37b21d 100644 --- a/gpustack_runtime/deployer/docker.py +++ b/gpustack_runtime/deployer/docker.py @@ -8,6 +8,7 @@ import socket import sys import tarfile +from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from math import ceil from pathlib import Path @@ -27,6 +28,7 @@ from .. import envs from ..logging import debug_log_exception, debug_log_warning from .__types__ import ( + DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS, Container, ContainerCheck, ContainerEnv, @@ -68,6 +70,9 @@ _LABEL_COMPONENT_NAME = f"{_LABEL_COMPONENT}-name" _LABEL_COMPONENT_INDEX = f"{_LABEL_COMPONENT}-index" _LABEL_COMPONENT_HEAL_PREFIX = f"{_LABEL_COMPONENT}-heal" +_LABEL_TERMINATION_GRACE_PERIOD_SECONDS = ( + f"{envs.GPUSTACK_RUNTIME_DEPLOY_LABEL_PREFIX}/termination-grace-period-seconds" +) @dataclass_json @@ -104,6 +109,8 @@ class DockerWorkloadPlan(WorkloadPlan): The group ID to own the filesystem of the workload. sysctls (dict[str, str] | None): Sysctls to set for the workload. + termination_grace_period_seconds (int): + Duration in seconds the containers of the workload need to terminate gracefully. containers (list[tuple[int, Container]] | None): List of containers in the workload. It must contain at least one "RUN" profile container. @@ -138,6 +145,12 @@ def validate_and_default(self): # Default and validate in the base class. super().validate_and_default() + # Carry the termination grace period on the containers, + # as the Docker models layer cannot express a create time stop timeout. + self.labels[_LABEL_TERMINATION_GRACE_PERIOD_SECONDS] = str( + self.termination_grace_period_seconds, + ) + # Adjust images. self.pause_image = adjust_image_with_envs(self.pause_image) self.unhealthy_restart_image = adjust_image_with_envs( @@ -1650,6 +1663,7 @@ def _delete( self, name: WorkloadName, namespace: WorkloadNamespace | None = None, + grace_period_seconds: int | None = None, ) -> WorkloadStatus | None: """ Delete a Docker workload. @@ -1659,6 +1673,9 @@ def _delete( The name of the workload. namespace: The namespace of the workload. + grace_period_seconds: + Duration in seconds the workload needs to terminate gracefully, + which overrides the one declared by the workload plan. Return: The status if found, None otherwise. @@ -1678,19 +1695,40 @@ def _delete( # Remove all containers with the workload label. try: d_containers = getattr(workload, "_d_containers", []) - # Remove non-pause containers first. + # Remove the unhealthy restart container first, + # otherwise it restarts the containers draining below. for c in d_containers: - if "-pause" not in c.name: + if c.labels.get(_LABEL_COMPONENT) == "unhealthy-restart": c.remove( force=True, ) - # Then remove pause containers. + # Then drain the non-pause containers concurrently, + # so each of them gets the whole grace period, + # and remove them. + d_drain_containers = [ + c + for c in d_containers + if c.labels.get(_LABEL_COMPONENT) not in ("pause", "unhealthy-restart") + ] + _stop_containers( + d_drain_containers, + _termination_grace_period_seconds(d_containers, grace_period_seconds), + ) + for c in d_drain_containers: + c.remove( + force=True, + ) + # Finally remove the pause containers, + # which hold the namespaces shared by the above containers. for c in d_containers: - if "-pause" in c.name: + if c.labels.get(_LABEL_COMPONENT) == "pause": c.remove( force=True, ) - except docker.errors.APIError as e: + # NB(thxCode): Deleting a workload always fails with an OperationError, + # as neither the Docker SDK errors nor the transport ones underneath them + # share a base class narrow enough to catch on its own. + except Exception as e: msg = f"Failed to delete containers for workload {name}{_detail_api_call_error(e)}" raise OperationError(msg) from e @@ -1711,7 +1749,7 @@ def _delete( v.remove( force=True, ) - except docker.errors.APIError as e: + except Exception as e: msg = f"Failed to delete volumes for workload {name}{_detail_api_call_error(e)}" raise OperationError(msg) from e @@ -2168,6 +2206,71 @@ def _has_restart_policy( ) +def _stop_containers( + containers: list[docker.models.containers.Container], + grace_period_seconds: int, +): + """ + Stop the given containers concurrently, + so each of them gets the whole grace period, + which mirrors how Kubernetes terminates the containers of a Pod. + + A container failing to stop is left to the forceful removal following this, + as the removal is the authoritative teardown. + + Args: + containers: + List of Docker containers to stop. + grace_period_seconds: + Duration in seconds the containers need to terminate gracefully. + + """ + if not containers: + return + + def stop(container: docker.models.containers.Container): + try: + container.stop( + timeout=grace_period_seconds, + ) + except Exception: + debug_log_exception(logger, f"Failed to stop container {container.name}") + + with ThreadPoolExecutor(max_workers=min(len(containers), 16)) as pool: + list(pool.map(stop, containers)) + + +def _termination_grace_period_seconds( + containers: list[docker.models.containers.Container], + override: int | None = None, +) -> int: + """ + Resolve how long the given containers may take to terminate gracefully, + which prefers the given override, then the one declared by the workload plan, + and finally the default. + + Args: + containers: + List of Docker containers in the workload. + override: + Duration in seconds overriding the one declared by the workload plan. + + Returns: + The duration in seconds. + + """ + if override is not None: + return override + + for c in containers: + declared = c.labels.get(_LABEL_TERMINATION_GRACE_PERIOD_SECONDS) + if declared: + with contextlib.suppress(ValueError): + return max(0, int(declared)) + + return DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS + + class DockerWorkloadExecStream(WorkloadExecStream): """ A WorkloadExecStream implementation for Docker exec socket streams. @@ -2202,14 +2305,14 @@ def close(self): self._sock.close() -def _detail_api_call_error(err: docker.errors.APIError) -> str: +def _detail_api_call_error(err: Exception) -> str: """ Explain a Docker API error in a concise way, if the envs.GPUSTACK_RUNTIME_DEPLOY_API_CALL_ERROR_DETAIL is enabled. Args: err: - The Docker API error. + The Docker API error, or the transport error underneath it. Returns: A concise explanation of the error. @@ -2218,6 +2321,9 @@ def _detail_api_call_error(err: docker.errors.APIError) -> str: if not envs.GPUSTACK_RUNTIME_DEPLOY_API_CALL_ERROR_DETAIL: return "" + if not isinstance(err, docker.errors.APIError): + return f": {err}" + msg = f": Docker {'Client' if err.is_client_error() else 'Server'} Error" if err.explanation: msg += f": {err.explanation}" diff --git a/gpustack_runtime/deployer/kuberentes.py b/gpustack_runtime/deployer/kuberentes.py index 72bbd59..fe85c2c 100644 --- a/gpustack_runtime/deployer/kuberentes.py +++ b/gpustack_runtime/deployer/kuberentes.py @@ -4,8 +4,10 @@ import json import logging import os +import re from dataclasses import dataclass, field from datetime import datetime, timezone +from decimal import Decimal from enum import Enum from pathlib import Path from typing import TYPE_CHECKING @@ -21,6 +23,7 @@ from ..logging import debug_log_exception from . import ContainerCheck, ContainerMountModeEnum, OperationError from .__types__ import ( + DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS, Container, ContainerPort, ContainerProfileEnum, @@ -135,6 +138,8 @@ class KubernetesWorkloadPlan(WorkloadPlan): The group ID to own the filesystem of the workload. sysctls (dict[str, str] | None): Sysctls to set for the workload. + termination_grace_period_seconds (int): + Duration in seconds the containers of the workload need to terminate gracefully. containers (list[tuple[int, Container]] | None): List of containers in the workload. It must contain at least one "RUN" profile container. @@ -1374,6 +1379,7 @@ def _create_pod( ), host_ipc=workload.host_ipc, share_process_namespace=workload.pid_shared, + termination_grace_period_seconds=workload.termination_grace_period_seconds, node_name=self._node_name, automount_service_account_token=False, volumes=( @@ -2202,6 +2208,7 @@ def _delete( self, name: WorkloadName, namespace: WorkloadNamespace | None = None, + grace_period_seconds: int | None = None, ) -> WorkloadStatus | None: """ Delete a Kubernetes workload. @@ -2211,6 +2218,9 @@ def _delete( The name of the workload. namespace: The namespace of the workload. + grace_period_seconds: + Duration in seconds the workload needs to terminate gracefully, + which overrides the one declared by the workload plan. Returns: The status if found, None otherwise. @@ -2241,6 +2251,7 @@ def _delete( namespace=namespace, label_selector=label_selector, propagation_policy=propagation_policy, + grace_period_seconds=grace_period_seconds, ) except kubernetes.client.exceptions.ApiException as e: if e.status != 405: @@ -2257,10 +2268,16 @@ def _delete( name=pod.metadata.name, namespace=namespace, propagation_policy=propagation_policy, + grace_period_seconds=grace_period_seconds, ) - except kubernetes.client.exceptions.ApiException as e2: + except Exception as e2: msg = f"Failed to delete pod of workload {name}{_detail_api_call_error(e2)}" raise OperationError(msg) from e2 + # NB(thxCode): Deleting a workload always fails with an OperationError, + # so the transport errors underneath the API errors do not escape either. + except Exception as e: + msg = f"Failed to delete pod of workload {name}{_detail_api_call_error(e)}" + raise OperationError(msg) from e # Remove all Services with the workload label. try: @@ -2287,9 +2304,14 @@ def _delete( namespace=namespace, propagation_policy=propagation_policy, ) - except kubernetes.client.exceptions.ApiException as e2: + except Exception as e2: msg = f"Failed to delete service of workload {name}{_detail_api_call_error(e2)}" raise OperationError(msg) from e2 + # NB(thxCode): Deleting a workload always fails with an OperationError, + # so the transport errors underneath the API errors do not escape either. + except Exception as e: + msg = f"Failed to delete service of workload {name}{_detail_api_call_error(e)}" + raise OperationError(msg) from e # Remove all ConfigMaps with the workload label. try: @@ -2314,9 +2336,14 @@ def _delete( namespace=namespace, propagation_policy=propagation_policy, ) - except kubernetes.client.exceptions.ApiException as e2: + except Exception as e2: msg = f"Failed to delete configmap of workload {name}{_detail_api_call_error(e2)}" raise OperationError(msg) from e2 + # NB(thxCode): Deleting a workload always fails with an OperationError, + # so the transport errors underneath the API errors do not escape either. + except Exception as e: + msg = f"Failed to delete configmap of workload {name}{_detail_api_call_error(e)}" + raise OperationError(msg) from e return workload @@ -2843,6 +2870,91 @@ def equal_services( return (aspec.ports or []) == (bspec.ports or []) +_RE_QUANTITY_NUMBER = re.compile( + r"^[+-]?([0-9]+(\.[0-9]*)?|\.[0-9]+)([eE][+-]?[0-9]+)?$", +) +""" +Regex for the number a Kubernetes resource quantity carries, +which accepts ASCII digits only. +""" + +_QUANTITY_SUFFIXES: dict[str, Decimal] = { + "Ki": Decimal(2) ** 10, + "Mi": Decimal(2) ** 20, + "Gi": Decimal(2) ** 30, + "Ti": Decimal(2) ** 40, + "Pi": Decimal(2) ** 50, + "Ei": Decimal(2) ** 60, + "n": Decimal(10) ** -9, + "u": Decimal(10) ** -6, + "m": Decimal(10) ** -3, + "k": Decimal(10) ** 3, + "M": Decimal(10) ** 6, + "G": Decimal(10) ** 9, + "T": Decimal(10) ** 12, + "P": Decimal(10) ** 15, + "E": Decimal(10) ** 18, +} +""" +Multipliers of the suffixes a Kubernetes resource quantity may carry. +""" + + +def _parse_quantity(value: float | str) -> Decimal | str: + """ + Parse a Kubernetes resource quantity into the number it denotes, + so the spellings the API server rewrites, e.g. "1.5Gi" into "1536Mi" + or 0.5 into "500m", still compare equal to what was requested. + + Args: + value: + The resource quantity to parse. + + Returns: + The number the quantity denotes, + or the value itself if it does not spell one. + + """ + text = str(value) + + number, multiplier = text, Decimal(1) + for suffix, suffix_multiplier in _QUANTITY_SUFFIXES.items(): + if text.endswith(suffix): + number, multiplier = text[: -len(suffix)], suffix_multiplier + break + + # NB(thxCode): Decimal is more permissive than Kubernetes, which accepts + # ASCII digits only, so an underscore separated or full-width spelling + # would otherwise read as the plain one + # and let an invalid declaration pass as an unchanged one. + if not _RE_QUANTITY_NUMBER.match(number): + return text + + return Decimal(number) * multiplier + + +def _container_resources( + container: kubernetes.client.V1Container, +) -> dict: + """ + Return the resources the given Container spec requests as numbers, + dropping the empty entries the API server fills in, + which an unset resources declaration does not carry. + + """ + if container.resources is None: + return {} + return { + kind: ( + {k: _parse_quantity(v) for k, v in values.items()} + if isinstance(values, dict) + else values + ) + for kind, values in container.resources.to_dict().items() + if values + } + + def equal_containers( a: kubernetes.client.V1Container, b: kubernetes.client.V1Container, @@ -2874,7 +2986,7 @@ def equal_containers( return False if (a.ports or []) != (b.ports or []): return False - if (a.resources or {}) != (b.resources or {}): + if _container_resources(a) != _container_resources(b): return False if (a.volume_mounts or []) != (b.volume_mounts or []): return False @@ -2893,6 +3005,19 @@ def equal_containers( return all(not (k not in benv or benv[k] != v) for k, v in aenv.items()) +def _pod_termination_grace_period_seconds( + spec: kubernetes.client.V1PodSpec, +) -> int: + """ + Return the termination grace period the given Pod spec settles on, + which is the API server default when the spec leaves it unset. + + """ + if spec.termination_grace_period_seconds is None: + return DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS + return spec.termination_grace_period_seconds + + def equal_pods( a: kubernetes.client.V1Pod, b: kubernetes.client.V1Pod, @@ -2923,14 +3048,20 @@ def equal_pods( return False if aspec.runtime_class_name != bspec.runtime_class_name: return False - if aspec.host_network != bspec.host_network: + # NB(thxCode): The API server drops the disabled toggles instead of + # echoing them back, so compare what they settle on, not what they carry. + if bool(aspec.host_network) != bool(bspec.host_network): return False - if aspec.host_ipc != bspec.host_ipc: + if bool(aspec.host_ipc) != bool(bspec.host_ipc): return False - if aspec.share_process_namespace != bspec.share_process_namespace: + if bool(aspec.share_process_namespace) != bool(bspec.share_process_namespace): return False if (aspec.restart_policy or "Always") != (bspec.restart_policy or "Always"): return False + if _pod_termination_grace_period_seconds( + aspec, + ) != _pod_termination_grace_period_seconds(bspec): + return False if aspec.node_name and bspec.node_name and aspec.node_name != bspec.node_name: return False if (aspec.volumes or []) != (bspec.volumes or []): @@ -3006,14 +3137,14 @@ def close(self): self._ws.close() -def _detail_api_call_error(err: kubernetes.client.exceptions.ApiException) -> str: +def _detail_api_call_error(err: Exception) -> str: """ Explain a Kubernetes API error in a concise way, if the envs.GPUSTACK_RUNTIME_DEPLOY_API_CALL_ERROR_DETAIL is enabled. Args: err: - The Kubernetes API error. + The Kubernetes API error, or the transport error underneath it. Returns: A concise explanation of the error. @@ -3022,6 +3153,9 @@ def _detail_api_call_error(err: kubernetes.client.exceptions.ApiException) -> st if not envs.GPUSTACK_RUNTIME_DEPLOY_API_CALL_ERROR_DETAIL: return "" + if not isinstance(err, kubernetes.client.exceptions.ApiException): + return f": {err}" + msg = ": Kubernetes Error" if err.reason: msg += f": {err.reason}" diff --git a/gpustack_runtime/deployer/podman.py b/gpustack_runtime/deployer/podman.py index 2279362..85cfdfa 100644 --- a/gpustack_runtime/deployer/podman.py +++ b/gpustack_runtime/deployer/podman.py @@ -8,6 +8,7 @@ import socket import sys import tarfile +from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from math import ceil from pathlib import Path @@ -30,6 +31,7 @@ from ..logging import debug_log_exception, debug_log_warning from .__patches__ import patch_render_payload from .__types__ import ( + DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS, Container, ContainerCheck, ContainerEnv, @@ -71,6 +73,9 @@ _LABEL_COMPONENT_NAME = f"{_LABEL_COMPONENT}-name" _LABEL_COMPONENT_INDEX = f"{_LABEL_COMPONENT}-index" _LABEL_COMPONENT_HEAL_PREFIX = f"{_LABEL_COMPONENT}-heal" +_LABEL_TERMINATION_GRACE_PERIOD_SECONDS = ( + f"{envs.GPUSTACK_RUNTIME_DEPLOY_LABEL_PREFIX}/termination-grace-period-seconds" +) @dataclass_json @@ -107,6 +112,8 @@ class PodmanWorkloadPlan(WorkloadPlan): The group ID to own the filesystem of the workload. sysctls (dict[str, str] | None): Sysctls to set for the workload. + termination_grace_period_seconds (int): + Duration in seconds the containers of the workload need to terminate gracefully. containers (list[tuple[int, Container]] | None): List of containers in the workload. It must contain at least one "RUN" profile container. @@ -141,6 +148,12 @@ def validate_and_default(self): # Default and validate in the base class. super().validate_and_default() + # Carry the termination grace period on the containers, + # as the Podman models layer cannot express a create time stop timeout. + self.labels[_LABEL_TERMINATION_GRACE_PERIOD_SECONDS] = str( + self.termination_grace_period_seconds, + ) + # Adjust images. self.pause_image = adjust_image_with_envs(self.pause_image) self.unhealthy_restart_image = adjust_image_with_envs( @@ -1593,6 +1606,7 @@ def _delete( self, name: WorkloadName, namespace: WorkloadNamespace | None = None, + grace_period_seconds: int | None = None, ) -> WorkloadStatus | None: """ Delete a Podman workload. @@ -1602,6 +1616,9 @@ def _delete( The name of the workload. namespace: The namespace of the workload. + grace_period_seconds: + Duration in seconds the workload needs to terminate gracefully, + which overrides the one declared by the workload plan. Return: The status if found, None otherwise. @@ -1621,19 +1638,40 @@ def _delete( # Remove all containers with the workload label. try: d_containers = getattr(workload, "_d_containers", []) - # Remove non-pause containers first. + # Remove the unhealthy restart container first, + # otherwise it restarts the containers draining below. for c in d_containers: - if "-pause" not in c.name: + if c.labels.get(_LABEL_COMPONENT) == "unhealthy-restart": c.remove( force=True, ) - # Then remove pause containers. + # Then drain the non-pause containers concurrently, + # so each of them gets the whole grace period, + # and remove them. + d_drain_containers = [ + c + for c in d_containers + if c.labels.get(_LABEL_COMPONENT) not in ("pause", "unhealthy-restart") + ] + _stop_containers( + d_drain_containers, + _termination_grace_period_seconds(d_containers, grace_period_seconds), + ) + for c in d_drain_containers: + c.remove( + force=True, + ) + # Finally remove the pause containers, + # which hold the namespaces shared by the above containers. for c in d_containers: - if "-pause" in c.name: + if c.labels.get(_LABEL_COMPONENT) == "pause": c.remove( force=True, ) - except podman.errors.APIError as e: + # NB(thxCode): Deleting a workload always fails with an OperationError, + # as neither the Podman SDK errors nor the transport ones underneath them + # share a base class narrow enough to catch on its own. + except Exception as e: msg = f"Failed to delete containers for workload {name}{_detail_api_call_error(e)}" raise OperationError(msg) from e @@ -1654,7 +1692,7 @@ def _delete( v.remove( force=True, ) - except podman.errors.APIError as e: + except Exception as e: msg = f"Failed to delete volumes for workload {name}{_detail_api_call_error(e)}" raise OperationError(msg) from e @@ -2112,6 +2150,74 @@ def _endoscopic_inspect(self) -> str: return safe_json(c_attrs, indent=2) +def _stop_containers( + containers: list[podman.domain.containers.Container], + grace_period_seconds: int, +): + """ + Stop the given containers concurrently, + so each of them gets the whole grace period, + which mirrors how Kubernetes terminates the containers of a Pod. + + A container failing to stop is left to the forceful removal following this, + as the removal is the authoritative teardown. + + Args: + containers: + List of Podman containers to stop. + grace_period_seconds: + Duration in seconds the containers need to terminate gracefully. + + """ + if not containers: + return + + def stop(container: podman.domain.containers.Container): + try: + container.stop( + timeout=grace_period_seconds, + # Tolerate the "already stopped" answer, + # which podman-py cannot decode on its own. + ignore=True, + ) + except Exception: + debug_log_exception(logger, f"Failed to stop container {container.name}") + + with ThreadPoolExecutor(max_workers=min(len(containers), 16)) as pool: + list(pool.map(stop, containers)) + + +def _termination_grace_period_seconds( + containers: list[podman.domain.containers.Container], + override: int | None = None, +) -> int: + """ + Resolve how long the given containers may take to terminate gracefully, + which prefers the given override, then the one declared by the workload plan, + and finally the default. + + Args: + containers: + List of Podman containers in the workload. + override: + Duration in seconds overriding the one declared by the workload plan. + + Returns: + The duration in seconds. + + """ + if override is not None: + return override + + for c in containers: + declared = c.labels.get(_LABEL_TERMINATION_GRACE_PERIOD_SECONDS) + if declared: + with contextlib.suppress(ValueError): + return max(0, int(declared)) + + return DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS + + def _has_restart_policy( container: podman.domain.containers.Container, ) -> bool: @@ -2154,14 +2260,14 @@ def close(self): self._sock.close() -def _detail_api_call_error(err: podman.errors.APIError) -> str: +def _detail_api_call_error(err: Exception) -> str: """ Explain a Podman API error in a concise way, if the envs.GPUSTACK_RUNTIME_DEPLOY_API_CALL_ERROR_DETAIL is enabled. Args: err: - The Podman API error. + The Podman API error, or the transport error underneath it. Returns: A concise explanation of the error. @@ -2170,6 +2276,9 @@ def _detail_api_call_error(err: podman.errors.APIError) -> str: if not envs.GPUSTACK_RUNTIME_DEPLOY_API_CALL_ERROR_DETAIL: return "" + if not isinstance(err, podman.errors.APIError): + return f": {err}" + msg = f": Podman {'Client' if err.is_client_error() else 'Server'} Error" if err.explanation: msg += f": {err.explanation}" diff --git a/tests/gpustack_runtime/deployer/test_termination_grace_period.py b/tests/gpustack_runtime/deployer/test_termination_grace_period.py new file mode 100644 index 0000000..f2e9323 --- /dev/null +++ b/tests/gpustack_runtime/deployer/test_termination_grace_period.py @@ -0,0 +1,728 @@ +# The cases below drive the deployers' own deletion path, which is where the +# graceful termination becomes observable without a live Docker daemon, Podman +# socket or Kubernetes cluster. +# ruff: noqa: SLF001 + +import threading +from types import SimpleNamespace + +import docker.errors +import kubernetes.client +import kubernetes.client.exceptions +import podman.errors +import pytest + +from gpustack_runtime.deployer.__types__ import ( + DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS, + Container, + ContainerProfileEnum, + Deployer, + OperationError, + WorkloadPlan, +) +from gpustack_runtime.deployer.docker import ( + _LABEL_COMPONENT as _DOCKER_LABEL_COMPONENT, +) +from gpustack_runtime.deployer.docker import ( + _LABEL_TERMINATION_GRACE_PERIOD_SECONDS as _DOCKER_LABEL_GRACE_PERIOD, +) +from gpustack_runtime.deployer.docker import ( + DockerDeployer, + DockerWorkloadPlan, +) +from gpustack_runtime.deployer.kuberentes import ( + KubernetesDeployer, + KubernetesWorkloadPlan, + equal_containers, + equal_pods, +) +from gpustack_runtime.deployer.podman import ( + _LABEL_COMPONENT as _PODMAN_LABEL_COMPONENT, +) +from gpustack_runtime.deployer.podman import ( + _LABEL_TERMINATION_GRACE_PERIOD_SECONDS as _PODMAN_LABEL_GRACE_PERIOD, +) +from gpustack_runtime.deployer.podman import ( + PodmanDeployer, + PodmanWorkloadPlan, +) + + +def _plan(**kwargs) -> WorkloadPlan: + return WorkloadPlan( + name="test", + labels={}, + containers=[ + Container( + name="run", + image="busybox:1.37", + profile=ContainerProfileEnum.RUN, + ), + ], + **kwargs, + ) + + +def _recorder() -> SimpleNamespace: + calls = [] + return SimpleNamespace( + calls=calls, + _delete=lambda *args, **kwargs: calls.append((args, kwargs)), + ) + + +def test_workload_plan_defaults_the_termination_grace_period(): + # An unset termination grace period falls back to the package default. + plan = _plan() + plan.validate_and_default() + + assert DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS == 30 + assert plan.termination_grace_period_seconds == 30 + + +def test_workload_plan_keeps_a_zero_termination_grace_period(): + # Zero means "kill immediately", it must survive defaulting. + plan = _plan(termination_grace_period_seconds=0) + plan.validate_and_default() + + assert plan.termination_grace_period_seconds == 0 + + +def test_workload_plan_defaults_a_none_termination_grace_period(): + # A JSON payload carrying an explicit null degrades to the default. + plan = _plan(termination_grace_period_seconds=None) + plan.validate_and_default() + + assert plan.termination_grace_period_seconds == 30 + + +def test_workload_plan_rejects_a_negative_termination_grace_period(): + plan = _plan(termination_grace_period_seconds=-1) + + with pytest.raises(ValueError, match="grace period"): + plan.validate_and_default() + + +def test_deployer_delete_forwards_the_grace_period(): + # The delete facade hands the grace period down to the deployer implementation. + rec = _recorder() + + Deployer.delete(rec, name="test", grace_period_seconds=5, async_mode=False) + + assert rec.calls == [(("test", None, 5), {})] + + +def test_deployer_delete_defaults_the_grace_period_to_none(): + # Without an explicit override, the implementation resolves the grace period itself. + rec = _recorder() + + Deployer.delete(rec, name="test", async_mode=False) + + assert rec.calls == [(("test", None, None), {})] + + +class _FakeContainer: + """ + A container recording the deletion calls it receives into a shared journal. + """ + + def __init__( + self, + journal: list, + name: str, + component: str, + component_label: str, + labels: dict[str, str] | None = None, + ): + self.name = name + self.labels = {**(labels or {}), component_label: component} + self.journal = journal + self.stop_error = None + self.stop_barrier = None + + def stop(self, **kwargs): + if self.stop_barrier: + # Only a container signalled while the others are still draining + # reaches the barrier, a serial drain times out on it instead. + self.stop_barrier.wait(timeout=5) + self.journal.append(("stop", self.name, kwargs)) + if self.stop_error: + raise self.stop_error + + def remove(self, **kwargs): + self.journal.append(("remove", self.name, kwargs)) + + +def _containers( + journal: list, + component_label: str, + labels: dict[str, str] | None = None, +) -> list: + return [ + _FakeContainer(journal, "test-pause", "pause", component_label, labels), + _FakeContainer(journal, "test-init-0", "init", component_label, labels), + _FakeContainer(journal, "test-run-1", "run", component_label, labels), + _FakeContainer( + journal, + "test-unhealthy-restart", + "unhealthy-restart", + component_label, + labels, + ), + ] + + +def _docker_containers(journal: list, labels: dict[str, str] | None = None) -> list: + return _containers(journal, _DOCKER_LABEL_COMPONENT, labels) + + +def _podman_containers(journal: list, labels: dict[str, str] | None = None) -> list: + return _containers(journal, _PODMAN_LABEL_COMPONENT, labels) + + +def _deployer(containers: list) -> SimpleNamespace: + workload = SimpleNamespace(_d_containers=containers) + return SimpleNamespace( + is_supported=lambda: True, + get=lambda **_kwargs: workload, + _client=SimpleNamespace( + volumes=SimpleNamespace(list=lambda **_kwargs: []), + ), + ) + + +_DRAINED = ("test-init-0", "test-run-1") + + +def _stops(journal: list) -> list: + # The drainable containers are stopped concurrently, + # so only the arguments are pinned, never the order between them. + return sorted(c for c in journal if c[0] == "stop") + + +def _assert_drained_before_removed(journal: list): + # Every drainable container must be signaled before any of them is removed, + # otherwise a slow container eats the grace period of the ones behind it. + last_stop = max(i for i, c in enumerate(journal) if c[0] == "stop") + first_remove = min( + i for i, c in enumerate(journal) if c[0] == "remove" and c[1] in _DRAINED + ) + assert last_stop < first_remove + + +def test_docker_delete_drains_the_workload_before_removing_it(): + # The unhealthy restart container goes first, otherwise it restarts the + # containers being drained; the pause container goes last, as it holds the + # namespaces the others share. + journal = [] + dep = _deployer(_docker_containers(journal)) + + DockerDeployer._delete(dep, name="test", grace_period_seconds=20) + + assert journal[0] == ("remove", "test-unhealthy-restart", {"force": True}) + assert journal[-1] == ("remove", "test-pause", {"force": True}) + _assert_drained_before_removed(journal) + assert _stops(journal) == [ + ("stop", "test-init-0", {"timeout": 20}), + ("stop", "test-run-1", {"timeout": 20}), + ] + + +def test_docker_delete_gives_every_container_the_whole_grace_period(): + # Mirrors Kubernetes: the grace period is not a budget shared between the + # containers, each of them gets all of it. + journal = [] + dep = _deployer(_docker_containers(journal)) + + DockerDeployer._delete(dep, name="test", grace_period_seconds=20) + + assert [c[2]["timeout"] for c in _stops(journal)] == [20, 20] + + +def test_docker_delete_reads_the_grace_period_from_the_container_label(): + # Without an explicit override, the grace period declared by the workload + # plan is read back from the container label. + journal = [] + dep = _deployer( + _docker_containers(journal, labels={_DOCKER_LABEL_GRACE_PERIOD: "45"}), + ) + + DockerDeployer._delete(dep, name="test") + + assert _stops(journal) == [ + ("stop", "test-init-0", {"timeout": 45}), + ("stop", "test-run-1", {"timeout": 45}), + ] + + +def test_docker_delete_falls_back_to_the_default_grace_period(): + # A workload created before the grace period existed carries no label. + journal = [] + dep = _deployer(_docker_containers(journal)) + + DockerDeployer._delete(dep, name="test") + + assert _stops(journal) == [ + ("stop", "test-init-0", {"timeout": 30}), + ("stop", "test-run-1", {"timeout": 30}), + ] + + +def test_docker_delete_kills_immediately_on_a_zero_grace_period(): + journal = [] + dep = _deployer( + _docker_containers(journal, labels={_DOCKER_LABEL_GRACE_PERIOD: "45"}), + ) + + DockerDeployer._delete(dep, name="test", grace_period_seconds=0) + + assert _stops(journal) == [ + ("stop", "test-init-0", {"timeout": 0}), + ("stop", "test-run-1", {"timeout": 0}), + ] + + +@pytest.mark.parametrize( + "deployer, make_containers", + [ + (DockerDeployer, _docker_containers), + (PodmanDeployer, _podman_containers), + ], +) +def test_delete_signals_the_containers_at_the_same_time(deployer, make_containers): + # Draining serially would leave the containers behind the first one with + # less than the grace period, which is the bug this branch exists to fix. + # Both deployers carry their own copy of the drain, so both are guarded. + journal = [] + containers = make_containers(journal) + barrier = threading.Barrier(len(_DRAINED)) + for c in containers: + if c.name in _DRAINED: + c.stop_barrier = barrier + dep = _deployer(containers) + + deployer._delete(dep, name="test", grace_period_seconds=20) + + assert len(_stops(journal)) == len(_DRAINED) + + +def test_podman_delete_survives_a_container_failing_to_stop(): + journal = [] + containers = _podman_containers(journal) + containers[2].stop_error = podman.errors.APIError("boom") + dep = _deployer(containers) + + PodmanDeployer._delete(dep, name="test", grace_period_seconds=20) + + assert journal[-1] == ("remove", "test-pause", {"force": True}) + assert ("remove", "test-run-1", {"force": True}) in journal + + +def test_docker_delete_survives_a_container_failing_to_stop(): + # The forceful removal is the authoritative teardown, a container refusing + # to drain must not strand the pause container and the volumes. + journal = [] + containers = _docker_containers(journal) + containers[2].stop_error = docker.errors.APIError("boom") + dep = _deployer(containers) + + DockerDeployer._delete(dep, name="test", grace_period_seconds=20) + + assert journal[-1] == ("remove", "test-pause", {"force": True}) + assert ("remove", "test-run-1", {"force": True}) in journal + + +def test_docker_workload_plan_stamps_the_grace_period_label(): + # The grace period must survive the create/delete round trip, as the Docker + # models layer cannot carry a create time stop timeout. + plan = DockerWorkloadPlan( + name="test", + termination_grace_period_seconds=30, + containers=[ + Container( + name="run", + image="busybox:1.37", + profile=ContainerProfileEnum.RUN, + ), + ], + ) + plan.validate_and_default() + + assert plan.labels[_DOCKER_LABEL_GRACE_PERIOD] == "30" + + +def test_docker_workload_plan_stamps_the_defaulted_grace_period_label(): + plan = DockerWorkloadPlan( + name="test", + termination_grace_period_seconds=None, + containers=[ + Container( + name="run", + image="busybox:1.37", + profile=ContainerProfileEnum.RUN, + ), + ], + ) + plan.validate_and_default() + + assert plan.labels[_DOCKER_LABEL_GRACE_PERIOD] == "30" + + +def test_podman_delete_drains_the_workload_before_removing_it(): + # Mirrors the Docker deletion path, but tolerates the "already stopped" + # answer, which podman-py cannot decode on its own. + journal = [] + dep = _deployer(_podman_containers(journal)) + + PodmanDeployer._delete(dep, name="test", grace_period_seconds=20) + + assert journal[0] == ("remove", "test-unhealthy-restart", {"force": True}) + assert journal[-1] == ("remove", "test-pause", {"force": True}) + _assert_drained_before_removed(journal) + assert _stops(journal) == [ + ("stop", "test-init-0", {"timeout": 20, "ignore": True}), + ("stop", "test-run-1", {"timeout": 20, "ignore": True}), + ] + + +def test_podman_delete_reads_the_grace_period_from_the_container_label(): + journal = [] + dep = _deployer( + _podman_containers(journal, labels={_PODMAN_LABEL_GRACE_PERIOD: "45"}), + ) + + PodmanDeployer._delete(dep, name="test") + + assert _stops(journal) == [ + ("stop", "test-init-0", {"timeout": 45, "ignore": True}), + ("stop", "test-run-1", {"timeout": 45, "ignore": True}), + ] + + +def test_podman_delete_falls_back_to_the_default_grace_period(): + journal = [] + dep = _deployer(_podman_containers(journal)) + + PodmanDeployer._delete(dep, name="test") + + assert _stops(journal) == [ + ("stop", "test-init-0", {"timeout": 30, "ignore": True}), + ("stop", "test-run-1", {"timeout": 30, "ignore": True}), + ] + + +def test_podman_delete_kills_immediately_on_a_zero_grace_period(): + journal = [] + dep = _deployer( + _podman_containers(journal, labels={_PODMAN_LABEL_GRACE_PERIOD: "45"}), + ) + + PodmanDeployer._delete(dep, name="test", grace_period_seconds=0) + + assert _stops(journal) == [ + ("stop", "test-init-0", {"timeout": 0, "ignore": True}), + ("stop", "test-run-1", {"timeout": 0, "ignore": True}), + ] + + +def test_podman_workload_plan_stamps_the_grace_period_label(): + plan = PodmanWorkloadPlan( + name="test", + termination_grace_period_seconds=30, + containers=[ + Container( + name="run", + image="busybox:1.37", + profile=ContainerProfileEnum.RUN, + ), + ], + ) + plan.validate_and_default() + + assert plan.labels[_PODMAN_LABEL_GRACE_PERIOD] == "30" + + +def test_podman_workload_plan_stamps_the_defaulted_grace_period_label(): + plan = PodmanWorkloadPlan( + name="test", + termination_grace_period_seconds=None, + containers=[ + Container( + name="run", + image="busybox:1.37", + profile=ContainerProfileEnum.RUN, + ), + ], + ) + plan.validate_and_default() + + assert plan.labels[_PODMAN_LABEL_GRACE_PERIOD] == "30" + + +def test_kubernetes_delete_reports_a_transport_failure_as_an_operation_error( + monkeypatch, +): + # A transport failure carries no HTTP status, so it must not escape the + # deletion as something other than an OperationError. + class _FailingCoreV1Api: + def __init__(self, client=None): + pass + + def delete_collection_namespaced_pod(self, **_kwargs): + msg = "connection reset" + raise OSError(msg) + + monkeypatch.setattr(kubernetes.client, "CoreV1Api", _FailingCoreV1Api) + dep = SimpleNamespace( + is_supported=lambda: True, + get=lambda **_kwargs: SimpleNamespace(name="test"), + _client=None, + ) + + with pytest.raises(OperationError): + KubernetesDeployer._delete(dep, name="test", namespace="default") + + +def test_kubernetes_pod_declares_the_termination_grace_period(monkeypatch): + # The Pod spec is the declarative source of truth on Kubernetes, replacing + # the API server default of 30 seconds. + monkeypatch.setattr( + "gpustack_runtime.deployer.kuberentes.get_resource_injection_policy", + lambda *_args: "env", + ) + monkeypatch.setattr( + "gpustack_runtime.deployer.kuberentes._resolve_runtime_class_name", + lambda *_args: None, + ) + + class _FakeCoreV1Api: + def __init__(self, client=None): + pass + + def read_namespaced_pod(self, name, namespace): + raise kubernetes.client.exceptions.ApiException(status=404) + + def create_namespaced_pod(self, namespace, body): + return body + + monkeypatch.setattr(kubernetes.client, "CoreV1Api", _FakeCoreV1Api) + + deployer = object.__new__(KubernetesDeployer) + Deployer.__init__(deployer, "test") + deployer._materials = {} + deployer._client = None + deployer._node_name = None + deployer._image_pull_secret = None + deployer._mutate_create_pod = lambda pod: pod + + workload = KubernetesWorkloadPlan( + name="test", + namespace="default", + termination_grace_period_seconds=30, + containers=[ + Container( + name="run", + image="busybox:1.37", + profile=ContainerProfileEnum.RUN, + ), + ], + ) + workload.validate_and_default() + pod = deployer._create_pod(workload, {}) + + assert pod.spec.termination_grace_period_seconds == 30 + + +class _FakeCoreV1DeleteApi: + """ + Stand-in for the Kubernetes core API, recording the deletion calls it receives. + """ + + def __init__(self, journal: list, collection_status: int | None = None): + self.journal = journal + self.collection_status = collection_status + + def __call__(self, client=None): + return self + + def delete_collection_namespaced_pod(self, **kwargs): + self.journal.append(("delete_collection_namespaced_pod", kwargs)) + if self.collection_status: + raise kubernetes.client.exceptions.ApiException( + status=self.collection_status, + ) + + def list_namespaced_pod(self, **_kwargs): + return SimpleNamespace( + items=[SimpleNamespace(metadata=SimpleNamespace(name="test"))], + ) + + def delete_namespaced_pod(self, **kwargs): + self.journal.append(("delete_namespaced_pod", kwargs)) + + def delete_collection_namespaced_service(self, **kwargs): + self.journal.append(("delete_collection_namespaced_service", kwargs)) + + def delete_collection_namespaced_config_map(self, **kwargs): + self.journal.append(("delete_collection_namespaced_config_map", kwargs)) + + +def _kubernetes_deployer(monkeypatch, journal: list, collection_status=None): + monkeypatch.setattr( + kubernetes.client, + "CoreV1Api", + _FakeCoreV1DeleteApi(journal, collection_status), + ) + return SimpleNamespace( + is_supported=lambda: True, + get=lambda **_kwargs: SimpleNamespace(name="test"), + _client=None, + ) + + +def test_kubernetes_delete_forwards_the_grace_period(monkeypatch): + journal = [] + dep = _kubernetes_deployer(monkeypatch, journal) + + KubernetesDeployer._delete( + dep, + name="test", + namespace="default", + grace_period_seconds=5, + ) + + pod_calls = [c for c in journal if "pod" in c[0]] + assert len(pod_calls) == 1 + assert pod_calls[0][1]["grace_period_seconds"] == 5 + # The grace period is meaningless for the other resources. + assert all("grace_period_seconds" not in c[1] for c in journal if "pod" not in c[0]) + + +def test_kubernetes_delete_omits_an_unset_grace_period(monkeypatch): + # Without an override, the Pod spec's own declaration applies. + journal = [] + dep = _kubernetes_deployer(monkeypatch, journal) + + KubernetesDeployer._delete(dep, name="test", namespace="default") + + pod_calls = [c for c in journal if "pod" in c[0]] + assert pod_calls[0][1]["grace_period_seconds"] is None + + +def test_kubernetes_delete_forwards_the_grace_period_on_the_fallback_path(monkeypatch): + # A cluster refusing collection deletion falls back to deleting Pod by Pod. + journal = [] + dep = _kubernetes_deployer(monkeypatch, journal, collection_status=405) + + KubernetesDeployer._delete( + dep, + name="test", + namespace="default", + grace_period_seconds=5, + ) + + fallback_calls = [c for c in journal if c[0] == "delete_namespaced_pod"] + assert len(fallback_calls) == 1 + assert fallback_calls[0][1]["name"] == "test" + assert fallback_calls[0][1]["grace_period_seconds"] == 5 + + +def _pod( + termination_grace_period_seconds=None, + host_network=None, + resources=None, +) -> kubernetes.client.V1Pod: + return kubernetes.client.V1Pod( + metadata=kubernetes.client.V1ObjectMeta(name="test"), + spec=kubernetes.client.V1PodSpec( + host_network=host_network, + termination_grace_period_seconds=termination_grace_period_seconds, + containers=[ + kubernetes.client.V1Container( + name="run", + image="busybox:1.37", + resources=resources, + ), + ], + ), + ) + + +def test_kubernetes_equal_pods_ignores_the_api_server_defaults(): + # The API server drops a disabled toggle and fills an empty resources + # declaration back in, neither of which is a change worth recreating for. + actual = _pod( + termination_grace_period_seconds=30, + host_network=None, + resources=kubernetes.client.V1ResourceRequirements(), + ) + desired = _pod( + termination_grace_period_seconds=30, + host_network=False, + resources=None, + ) + + assert equal_pods(actual, desired) + + +def test_kubernetes_equal_pods_treats_an_unset_grace_period_as_the_default(): + # A Pod created before the grace period existed carries the API server + # default, which is what the plan default settles on too. + actual = _pod(termination_grace_period_seconds=None) + desired = _pod(termination_grace_period_seconds=30) + + assert equal_pods(actual, desired) + + +def test_kubernetes_equal_pods_detects_a_changed_grace_period(): + actual = _pod(termination_grace_period_seconds=30) + desired = _pod(termination_grace_period_seconds=10) + + assert not equal_pods(actual, desired) + + +def _container(cpu, memory) -> kubernetes.client.V1Container: + return kubernetes.client.V1Container( + name="run", + image="busybox:1.37", + resources=kubernetes.client.V1ResourceRequirements( + limits={"cpu": cpu, "memory": memory}, + requests={"cpu": cpu, "memory": memory}, + ), + ) + + +@pytest.mark.parametrize( + "declared, stored", + [ + # The API server rewrites a fractional quantity into its milli form. + ("0.5", "500m"), + # And a suffixed quantity into the largest suffix dividing it exactly. + ("1.5Gi", "1536Mi"), + # While an already canonical quantity is kept verbatim. + ("1Gi", "1Gi"), + ("1073741824", "1073741824"), + ], +) +def test_kubernetes_equal_containers_reads_a_rewritten_quantity(declared, stored): + # The API server stores a quantity in its own spelling, which must not read + # as a change, or the Pod is recreated on every deployment. + assert equal_containers(_container(stored, stored), _container(declared, declared)) + + +def test_kubernetes_equal_containers_detects_a_changed_quantity(): + assert not equal_containers(_container("1", "1Gi"), _container("2", "1Gi")) + assert not equal_containers(_container("1", "1Gi"), _container("1", "2Gi")) + + +def test_kubernetes_equal_containers_keeps_a_non_quantity_verbatim(): + # Device resources carry values that do not spell a quantity. + assert equal_containers(_container("1", "all"), _container("1", "all")) + assert not equal_containers(_container("1", "all"), _container("1", "0,1")) + + +@pytest.mark.parametrize("declared", ["1_0m", "\uff11\uff10m", "0x10", " 10m"]) +def test_kubernetes_equal_containers_refuses_a_non_ascii_quantity(declared): + # Kubernetes accepts ASCII digits only, so a declaration Python would read + # as a number must not pass as an unchanged one. + assert not equal_containers(_container("1", "10m"), _container("1", declared))