From b2122f9aea22a78efbdf29bdfb7907b5d06a40ea Mon Sep 17 00:00:00 2001 From: thxCode Date: Wed, 19 Aug 2026 10:46:30 +0800 Subject: [PATCH 1/7] feat(deployer): introduce workload termination grace period - Add `DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS`, defaulting to 15 seconds. - Declare `WorkloadPlan.termination_grace_period_seconds`, which defaults an unset value and rejects a negative one. - Accept `grace_period_seconds` on workload deletion, which overrides the duration declared by the workload plan. Task 1 of graceful-workload-termination. Signed-off-by: thxCode --- gpustack_runtime/deployer/__init__.py | 10 ++- gpustack_runtime/deployer/__types__.py | 33 +++++++- gpustack_runtime/deployer/docker.py | 4 + gpustack_runtime/deployer/kuberentes.py | 4 + gpustack_runtime/deployer/podman.py | 4 + .../deployer/test_termination_grace_period.py | 84 +++++++++++++++++++ 6 files changed, 137 insertions(+), 2 deletions(-) create mode 100644 tests/gpustack_runtime/deployer/test_termination_grace_period.py diff --git a/gpustack_runtime/deployer/__init__.py b/gpustack_runtime/deployer/__init__.py index 7a1d39a..306daf1 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,6 +160,9 @@ 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. @@ -174,7 +178,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..57118b6 100644 --- a/gpustack_runtime/deployer/__types__.py +++ b/gpustack_runtime/deployer/__types__.py @@ -832,6 +832,11 @@ class WorkloadSecurity: Name for a workload. """ +DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS = 15 +""" +Default duration in seconds a workload needs to terminate gracefully. +""" + @dataclass_json @dataclass @@ -862,6 +867,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. @@ -904,6 +911,12 @@ class WorkloadPlan(WorkloadSecurity): """ Configure shared memory size for the workload. """ + 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. + """ containers: list[Container] | None = None """ Containers in the workload. @@ -957,6 +970,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 +2131,7 @@ def delete( self, name: WorkloadName, namespace: WorkloadNamespace | None = None, + grace_period_seconds: int | None = None, async_mode: bool | None = None, ) -> WorkloadStatus | None: """ @@ -2119,6 +2142,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. @@ -2138,6 +2164,7 @@ def delete( self._delete, name, namespace, + grace_period_seconds, ) return future.result() except OperationError: @@ -2146,13 +2173,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 +2190,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..110682d 100644 --- a/gpustack_runtime/deployer/docker.py +++ b/gpustack_runtime/deployer/docker.py @@ -1650,6 +1650,7 @@ def _delete( self, name: WorkloadName, namespace: WorkloadNamespace | None = None, + grace_period_seconds: int | None = None, ) -> WorkloadStatus | None: """ Delete a Docker workload. @@ -1659,6 +1660,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/kuberentes.py b/gpustack_runtime/deployer/kuberentes.py index 72bbd59..77d71d4 100644 --- a/gpustack_runtime/deployer/kuberentes.py +++ b/gpustack_runtime/deployer/kuberentes.py @@ -2202,6 +2202,7 @@ def _delete( self, name: WorkloadName, namespace: WorkloadNamespace | None = None, + grace_period_seconds: int | None = None, ) -> WorkloadStatus | None: """ Delete a Kubernetes workload. @@ -2211,6 +2212,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. diff --git a/gpustack_runtime/deployer/podman.py b/gpustack_runtime/deployer/podman.py index 2279362..f3827e0 100644 --- a/gpustack_runtime/deployer/podman.py +++ b/gpustack_runtime/deployer/podman.py @@ -1593,6 +1593,7 @@ def _delete( self, name: WorkloadName, namespace: WorkloadNamespace | None = None, + grace_period_seconds: int | None = None, ) -> WorkloadStatus | None: """ Delete a Podman workload. @@ -1602,6 +1603,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/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..eb363df --- /dev/null +++ b/tests/gpustack_runtime/deployer/test_termination_grace_period.py @@ -0,0 +1,84 @@ +from types import SimpleNamespace + +import pytest + +from gpustack_runtime.deployer.__types__ import ( + DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS, + Container, + ContainerProfileEnum, + Deployer, + WorkloadPlan, +) + + +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 == 15 + assert plan.termination_grace_period_seconds == 15 + + +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 == 15 + + +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), {})] From a5acbfbadbde4d2b33da99978f060b5e1f6a658c Mon Sep 17 00:00:00 2001 From: thxCode Date: Wed, 19 Aug 2026 10:51:58 +0800 Subject: [PATCH 2/7] fix(deployer): drain docker workload before removing it - Stamp the termination grace period on the containers, as the Docker models layer cannot express a create time stop timeout. - Remove the unhealthy restart container first, otherwise it restarts the containers being drained. - Stop the non-pause containers within the grace period shared by all of them, then remove them, and remove the pause containers at last. Task 2 of graceful-workload-termination. Signed-off-by: thxCode --- gpustack_runtime/deployer/docker.py | 67 +++++++- .../deployer/test_termination_grace_period.py | 156 ++++++++++++++++++ 2 files changed, 219 insertions(+), 4 deletions(-) diff --git a/gpustack_runtime/deployer/docker.py b/gpustack_runtime/deployer/docker.py index 110682d..d056547 100644 --- a/gpustack_runtime/deployer/docker.py +++ b/gpustack_runtime/deployer/docker.py @@ -8,6 +8,7 @@ import socket import sys import tarfile +import time 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 @@ -138,6 +143,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( @@ -1682,15 +1693,32 @@ 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 within the grace period, + # which is shared by all of them, and remove them. + deadline = time.monotonic() + _termination_grace_period_seconds( + d_containers, + grace_period_seconds, + ) + for c in d_containers: + if c.labels.get(_LABEL_COMPONENT) in ("pause", "unhealthy-restart"): + continue + c.stop( + timeout=max(0, ceil(deadline - time.monotonic())), + ) + 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, ) @@ -2172,6 +2200,37 @@ def _has_restart_policy( ) +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 max(0, 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. diff --git a/tests/gpustack_runtime/deployer/test_termination_grace_period.py b/tests/gpustack_runtime/deployer/test_termination_grace_period.py index eb363df..99d0056 100644 --- a/tests/gpustack_runtime/deployer/test_termination_grace_period.py +++ b/tests/gpustack_runtime/deployer/test_termination_grace_period.py @@ -1,3 +1,8 @@ +# 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 + from types import SimpleNamespace import pytest @@ -9,6 +14,16 @@ Deployer, 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, +) def _plan(**kwargs) -> WorkloadPlan: @@ -82,3 +97,144 @@ def test_deployer_delete_defaults_the_grace_period_to_none(): 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, + labels: dict[str, str] | None = None, + ): + self.name = name + self.labels = {**(labels or {}), _DOCKER_LABEL_COMPONENT: component} + self.journal = journal + + def stop(self, **kwargs): + self.journal.append(("stop", self.name, kwargs)) + + def remove(self, **kwargs): + self.journal.append(("remove", self.name, kwargs)) + + +def _docker_containers(journal: list, labels: dict[str, str] | None = None) -> list: + return [ + _FakeContainer(journal, "test-pause", "pause", labels), + _FakeContainer(journal, "test-init-0", "init", labels), + _FakeContainer(journal, "test-run-1", "run", labels), + _FakeContainer(journal, "test-unhealthy-restart", "unhealthy-restart", labels), + ] + + +def _docker_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: []), + ), + ) + + +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 = _docker_deployer(_docker_containers(journal)) + + DockerDeployer._delete(dep, name="test", grace_period_seconds=20) + + assert journal == [ + ("remove", "test-unhealthy-restart", {"force": True}), + ("stop", "test-init-0", {"timeout": 20}), + ("remove", "test-init-0", {"force": True}), + ("stop", "test-run-1", {"timeout": 20}), + ("remove", "test-run-1", {"force": True}), + ("remove", "test-pause", {"force": True}), + ] + + +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 = _docker_deployer( + _docker_containers(journal, labels={_DOCKER_LABEL_GRACE_PERIOD: "30"}), + ) + + DockerDeployer._delete(dep, name="test") + + assert [c for c in journal if c[0] == "stop"] == [ + ("stop", "test-init-0", {"timeout": 30}), + ("stop", "test-run-1", {"timeout": 30}), + ] + + +def test_docker_delete_falls_back_to_the_default_grace_period(): + # A workload created before the grace period existed carries no label. + journal = [] + dep = _docker_deployer(_docker_containers(journal)) + + DockerDeployer._delete(dep, name="test") + + assert [c for c in journal if c[0] == "stop"] == [ + ("stop", "test-init-0", {"timeout": 15}), + ("stop", "test-run-1", {"timeout": 15}), + ] + + +def test_docker_delete_kills_immediately_on_a_zero_grace_period(): + journal = [] + dep = _docker_deployer( + _docker_containers(journal, labels={_DOCKER_LABEL_GRACE_PERIOD: "30"}), + ) + + DockerDeployer._delete(dep, name="test", grace_period_seconds=0) + + assert [c for c in journal if c[0] == "stop"] == [ + ("stop", "test-init-0", {"timeout": 0}), + ("stop", "test-run-1", {"timeout": 0}), + ] + + +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] == "15" From 799b3672b36433be55a32365291d0cd7525617a1 Mon Sep 17 00:00:00 2001 From: thxCode Date: Wed, 19 Aug 2026 11:02:32 +0800 Subject: [PATCH 3/7] fix(deployer): drain podman workload before removing it - Stamp the termination grace period on the containers, as the Podman models layer cannot express a create time stop timeout. - Remove the unhealthy restart container first, otherwise it restarts the containers being drained. - Stop the non-pause containers within the grace period shared by all of them, then remove them, and remove the pause containers at last. - Ignore the "already stopped" answer, which podman-py cannot decode on its own. Task 3 of graceful-workload-termination. Signed-off-by: thxCode --- gpustack_runtime/deployer/podman.py | 70 +++++++++- .../deployer/test_termination_grace_period.py | 126 ++++++++++++++++-- 2 files changed, 181 insertions(+), 15 deletions(-) diff --git a/gpustack_runtime/deployer/podman.py b/gpustack_runtime/deployer/podman.py index f3827e0..e7ccfbc 100644 --- a/gpustack_runtime/deployer/podman.py +++ b/gpustack_runtime/deployer/podman.py @@ -8,6 +8,7 @@ import socket import sys import tarfile +import time 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 @@ -141,6 +146,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( @@ -1625,15 +1636,35 @@ 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 within the grace period, + # which is shared by all of them, and remove them. + deadline = time.monotonic() + _termination_grace_period_seconds( + d_containers, + grace_period_seconds, + ) + for c in d_containers: + if c.labels.get(_LABEL_COMPONENT) in ("pause", "unhealthy-restart"): + continue + c.stop( + timeout=max(0, ceil(deadline - time.monotonic())), + # Tolerate the "already stopped" answer, + # which podman-py cannot decode on its own. + ignore=True, + ) + 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, ) @@ -2116,6 +2147,37 @@ def _endoscopic_inspect(self) -> str: return safe_json(c_attrs, indent=2) +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 max(0, 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: diff --git a/tests/gpustack_runtime/deployer/test_termination_grace_period.py b/tests/gpustack_runtime/deployer/test_termination_grace_period.py index 99d0056..3a9152b 100644 --- a/tests/gpustack_runtime/deployer/test_termination_grace_period.py +++ b/tests/gpustack_runtime/deployer/test_termination_grace_period.py @@ -24,6 +24,16 @@ DockerDeployer, DockerWorkloadPlan, ) +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: @@ -109,10 +119,11 @@ def __init__( journal: list, name: str, component: str, + component_label: str, labels: dict[str, str] | None = None, ): self.name = name - self.labels = {**(labels or {}), _DOCKER_LABEL_COMPONENT: component} + self.labels = {**(labels or {}), component_label: component} self.journal = journal def stop(self, **kwargs): @@ -122,16 +133,34 @@ def remove(self, **kwargs): self.journal.append(("remove", self.name, kwargs)) -def _docker_containers(journal: list, labels: dict[str, str] | None = None) -> list: +def _containers( + journal: list, + component_label: str, + labels: dict[str, str] | None = None, +) -> list: return [ - _FakeContainer(journal, "test-pause", "pause", labels), - _FakeContainer(journal, "test-init-0", "init", labels), - _FakeContainer(journal, "test-run-1", "run", labels), - _FakeContainer(journal, "test-unhealthy-restart", "unhealthy-restart", labels), + _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_deployer(containers: list) -> SimpleNamespace: +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, @@ -147,7 +176,7 @@ def test_docker_delete_drains_the_workload_before_removing_it(): # containers being drained; the pause container goes last, as it holds the # namespaces the others share. journal = [] - dep = _docker_deployer(_docker_containers(journal)) + dep = _deployer(_docker_containers(journal)) DockerDeployer._delete(dep, name="test", grace_period_seconds=20) @@ -165,7 +194,7 @@ 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 = _docker_deployer( + dep = _deployer( _docker_containers(journal, labels={_DOCKER_LABEL_GRACE_PERIOD: "30"}), ) @@ -180,7 +209,7 @@ def test_docker_delete_reads_the_grace_period_from_the_container_label(): def test_docker_delete_falls_back_to_the_default_grace_period(): # A workload created before the grace period existed carries no label. journal = [] - dep = _docker_deployer(_docker_containers(journal)) + dep = _deployer(_docker_containers(journal)) DockerDeployer._delete(dep, name="test") @@ -192,7 +221,7 @@ def test_docker_delete_falls_back_to_the_default_grace_period(): def test_docker_delete_kills_immediately_on_a_zero_grace_period(): journal = [] - dep = _docker_deployer( + dep = _deployer( _docker_containers(journal, labels={_DOCKER_LABEL_GRACE_PERIOD: "30"}), ) @@ -238,3 +267,78 @@ def test_docker_workload_plan_stamps_the_defaulted_grace_period_label(): plan.validate_and_default() assert plan.labels[_DOCKER_LABEL_GRACE_PERIOD] == "15" + + +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 == [ + ("remove", "test-unhealthy-restart", {"force": True}), + ("stop", "test-init-0", {"timeout": 20, "ignore": True}), + ("remove", "test-init-0", {"force": True}), + ("stop", "test-run-1", {"timeout": 20, "ignore": True}), + ("remove", "test-run-1", {"force": True}), + ("remove", "test-pause", {"force": True}), + ] + + +def test_podman_delete_reads_the_grace_period_from_the_container_label(): + journal = [] + dep = _deployer( + _podman_containers(journal, labels={_PODMAN_LABEL_GRACE_PERIOD: "30"}), + ) + + PodmanDeployer._delete(dep, name="test") + + assert [c for c in journal if c[0] == "stop"] == [ + ("stop", "test-init-0", {"timeout": 30, "ignore": True}), + ("stop", "test-run-1", {"timeout": 30, "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 [c for c in journal if c[0] == "stop"] == [ + ("stop", "test-init-0", {"timeout": 15, "ignore": True}), + ("stop", "test-run-1", {"timeout": 15, "ignore": True}), + ] + + +def test_podman_delete_kills_immediately_on_a_zero_grace_period(): + journal = [] + dep = _deployer( + _podman_containers(journal, labels={_PODMAN_LABEL_GRACE_PERIOD: "30"}), + ) + + PodmanDeployer._delete(dep, name="test", grace_period_seconds=0) + + assert [c for c in journal if c[0] == "stop"] == [ + ("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" From 2fddfab4d56d4b7ffcc4b222689706e235b4f52c Mon Sep 17 00:00:00 2001 From: thxCode Date: Wed, 19 Aug 2026 11:08:31 +0800 Subject: [PATCH 4/7] feat(deployer): declare and override kubernetes termination grace period - Declare the termination grace period on the Pod spec, replacing the API server default. - Forward the overriding grace period when deleting Pods, both by collection and one by one. Task 4 of graceful-workload-termination. Signed-off-by: thxCode --- gpustack_runtime/deployer/kuberentes.py | 3 + .../deployer/test_termination_grace_period.py | 150 ++++++++++++++++++ 2 files changed, 153 insertions(+) diff --git a/gpustack_runtime/deployer/kuberentes.py b/gpustack_runtime/deployer/kuberentes.py index 77d71d4..158d04e 100644 --- a/gpustack_runtime/deployer/kuberentes.py +++ b/gpustack_runtime/deployer/kuberentes.py @@ -1374,6 +1374,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=( @@ -2245,6 +2246,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: @@ -2261,6 +2263,7 @@ 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: msg = f"Failed to delete pod of workload {name}{_detail_api_call_error(e2)}" diff --git a/tests/gpustack_runtime/deployer/test_termination_grace_period.py b/tests/gpustack_runtime/deployer/test_termination_grace_period.py index 3a9152b..028ba94 100644 --- a/tests/gpustack_runtime/deployer/test_termination_grace_period.py +++ b/tests/gpustack_runtime/deployer/test_termination_grace_period.py @@ -5,6 +5,8 @@ from types import SimpleNamespace +import kubernetes.client +import kubernetes.client.exceptions import pytest from gpustack_runtime.deployer.__types__ import ( @@ -24,6 +26,10 @@ DockerDeployer, DockerWorkloadPlan, ) +from gpustack_runtime.deployer.kuberentes import ( + KubernetesDeployer, + KubernetesWorkloadPlan, +) from gpustack_runtime.deployer.podman import ( _LABEL_COMPONENT as _PODMAN_LABEL_COMPONENT, ) @@ -342,3 +348,147 @@ def test_podman_workload_plan_stamps_the_grace_period_label(): plan.validate_and_default() assert plan.labels[_PODMAN_LABEL_GRACE_PERIOD] == "30" + + +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 From 50102b1c26f783ccf856e9f3906a6cf7acb9d8a8 Mon Sep 17 00:00:00 2001 From: thxCode Date: Wed, 19 Aug 2026 11:10:33 +0800 Subject: [PATCH 5/7] feat(cmds): expose the termination grace period on workload deletion - Accept `--grace-period-seconds` on both `delete` and `delete-all`, which overrides the duration declared by the workload plan. Task 5 of graceful-workload-termination. Signed-off-by: thxCode --- gpustack_runtime/cmds/deployer.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/gpustack_runtime/cmds/deployer.py b/gpustack_runtime/cmds/deployer.py index e473e5e..473af43 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 @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,19 @@ 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", + ) + 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 +461,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: From 6dc87eab5f7fb8d4173db4e4af3f8930aa4ac536 Mon Sep 17 00:00:00 2001 From: thxCode Date: Wed, 19 Aug 2026 11:55:29 +0800 Subject: [PATCH 6/7] fix(deployer): give every container the whole termination grace period - Stop the drainable containers concurrently instead of walking them against one shared deadline, so a container ignoring the signal no longer starves the ones behind it, which is how Kubernetes terminates a Pod. - Leave a container failing to stop to the forceful removal following it, instead of aborting the deletion and stranding the pause container. - Reject a negative overriding grace period, which used to be clamped to an immediate kill by Docker and Podman, but forwarded verbatim to Kubernetes. - Report every workload deletion failure as an operation error, as neither the SDK errors nor the transport ones underneath share a narrow base class. - Default the grace period to 30 seconds, aligning with the Kubernetes API server, and compare it when reconciling a Pod. - Stop recreating an unchanged Pod: the API server drops a disabled toggle and fills an empty resources declaration back in, neither being a real change. - Declare the grace period after the containers, so positional construction of a workload plan keeps binding the container list to the containers field. - Accept an unset grace period on the deletion commands, and document that deleting all workloads may take the grace period per workload. Review follow-ups of graceful-workload-termination. Signed-off-by: thxCode --- gpustack_runtime/cmds/deployer.py | 6 +- gpustack_runtime/deployer/__init__.py | 2 + gpustack_runtime/deployer/__types__.py | 21 ++- gpustack_runtime/deployer/docker.py | 77 ++++++-- gpustack_runtime/deployer/kuberentes.py | 44 ++++- gpustack_runtime/deployer/podman.py | 83 ++++++--- .../deployer/test_termination_grace_period.py | 165 ++++++++++++++---- 7 files changed, 316 insertions(+), 82 deletions(-) diff --git a/gpustack_runtime/cmds/deployer.py b/gpustack_runtime/cmds/deployer.py index 473af43..d626808 100644 --- a/gpustack_runtime/cmds/deployer.py +++ b/gpustack_runtime/cmds/deployer.py @@ -362,7 +362,7 @@ class DeleteWorkloadSubCommand(SubCommand): namespace: str name: str - grace_period_seconds: int + grace_period_seconds: int | None @staticmethod def register(parser: _SubParsersAction): @@ -442,7 +442,9 @@ def register(parser: _SubParsersAction): "--grace-period-seconds", type=int, help="Duration in seconds the workloads need to terminate gracefully, " - "which overrides the one declared by the workload plans", + "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) diff --git a/gpustack_runtime/deployer/__init__.py b/gpustack_runtime/deployer/__init__.py index 306daf1..ed12fd8 100644 --- a/gpustack_runtime/deployer/__init__.py +++ b/gpustack_runtime/deployer/__init__.py @@ -168,6 +168,8 @@ def delete_workload( The status if found, None otherwise. Raises: + ValueError: + If the grace period is negative. UnsupportedError: If no deployer supports the given workload. OperationError: diff --git a/gpustack_runtime/deployer/__types__.py b/gpustack_runtime/deployer/__types__.py index 57118b6..d15c215 100644 --- a/gpustack_runtime/deployer/__types__.py +++ b/gpustack_runtime/deployer/__types__.py @@ -832,9 +832,10 @@ class WorkloadSecurity: Name for a workload. """ -DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS = 15 +DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS = 30 """ -Default duration in seconds a workload needs to terminate gracefully. +Default duration in seconds a workload needs to terminate gracefully, +which aligns with the Kubernetes API server default. """ @@ -911,17 +912,17 @@ class WorkloadPlan(WorkloadSecurity): """ Configure shared memory size for the workload. """ + containers: list[Container] | None = None + """ + 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. """ - containers: list[Container] | None = None - """ - Containers in the workload. - It must contain at least one "RUN" profile container. - """ ## ## The below are internal use only fields. @@ -2154,10 +2155,16 @@ 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( diff --git a/gpustack_runtime/deployer/docker.py b/gpustack_runtime/deployer/docker.py index d056547..50a5cae 100644 --- a/gpustack_runtime/deployer/docker.py +++ b/gpustack_runtime/deployer/docker.py @@ -8,7 +8,7 @@ import socket import sys import tarfile -import time +from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from math import ceil from pathlib import Path @@ -109,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. @@ -1700,18 +1702,19 @@ def _delete( c.remove( force=True, ) - # Then drain the non-pause containers within the grace period, - # which is shared by all of them, and remove them. - deadline = time.monotonic() + _termination_grace_period_seconds( - d_containers, - grace_period_seconds, + # 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_containers: - if c.labels.get(_LABEL_COMPONENT) in ("pause", "unhealthy-restart"): - continue - c.stop( - timeout=max(0, ceil(deadline - time.monotonic())), - ) + for c in d_drain_containers: c.remove( force=True, ) @@ -1722,7 +1725,10 @@ def _delete( 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 @@ -1743,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 @@ -2200,6 +2206,40 @@ 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=len(containers)) as pool: + list(pool.map(stop, containers)) + + def _termination_grace_period_seconds( containers: list[docker.models.containers.Container], override: int | None = None, @@ -2220,7 +2260,7 @@ def _termination_grace_period_seconds( """ if override is not None: - return max(0, override) + return override for c in containers: declared = c.labels.get(_LABEL_TERMINATION_GRACE_PERIOD_SECONDS) @@ -2265,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. @@ -2281,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 158d04e..0a301e1 100644 --- a/gpustack_runtime/deployer/kuberentes.py +++ b/gpustack_runtime/deployer/kuberentes.py @@ -21,6 +21,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 +136,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. @@ -2850,6 +2853,20 @@ def equal_services( return (aspec.ports or []) == (bspec.ports or []) +def _container_resources( + container: kubernetes.client.V1Container, +) -> dict: + """ + Return the resources the given Container spec requests, + dropping the empty entries the API server fills in, + which an unset resources declaration does not carry. + + """ + if container.resources is None: + return {} + return {k: v for k, v in container.resources.to_dict().items() if v} + + def equal_containers( a: kubernetes.client.V1Container, b: kubernetes.client.V1Container, @@ -2881,7 +2898,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 @@ -2900,6 +2917,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, @@ -2930,14 +2960,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 []): diff --git a/gpustack_runtime/deployer/podman.py b/gpustack_runtime/deployer/podman.py index e7ccfbc..27306d6 100644 --- a/gpustack_runtime/deployer/podman.py +++ b/gpustack_runtime/deployer/podman.py @@ -8,7 +8,7 @@ import socket import sys import tarfile -import time +from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from math import ceil from pathlib import Path @@ -112,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. @@ -1643,21 +1645,19 @@ def _delete( c.remove( force=True, ) - # Then drain the non-pause containers within the grace period, - # which is shared by all of them, and remove them. - deadline = time.monotonic() + _termination_grace_period_seconds( - d_containers, - grace_period_seconds, + # 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_containers: - if c.labels.get(_LABEL_COMPONENT) in ("pause", "unhealthy-restart"): - continue - c.stop( - timeout=max(0, ceil(deadline - time.monotonic())), - # Tolerate the "already stopped" answer, - # which podman-py cannot decode on its own. - ignore=True, - ) + for c in d_drain_containers: c.remove( force=True, ) @@ -1668,7 +1668,10 @@ def _delete( 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 @@ -1689,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 @@ -2147,6 +2150,43 @@ 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=len(containers)) as pool: + list(pool.map(stop, containers)) + + def _termination_grace_period_seconds( containers: list[podman.domain.containers.Container], override: int | None = None, @@ -2167,7 +2207,7 @@ def _termination_grace_period_seconds( """ if override is not None: - return max(0, override) + return override for c in containers: declared = c.labels.get(_LABEL_TERMINATION_GRACE_PERIOD_SECONDS) @@ -2220,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. @@ -2236,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 index 028ba94..c02fd75 100644 --- a/tests/gpustack_runtime/deployer/test_termination_grace_period.py +++ b/tests/gpustack_runtime/deployer/test_termination_grace_period.py @@ -5,6 +5,7 @@ from types import SimpleNamespace +import docker.errors import kubernetes.client import kubernetes.client.exceptions import pytest @@ -29,6 +30,7 @@ from gpustack_runtime.deployer.kuberentes import ( KubernetesDeployer, KubernetesWorkloadPlan, + equal_pods, ) from gpustack_runtime.deployer.podman import ( _LABEL_COMPONENT as _PODMAN_LABEL_COMPONENT, @@ -70,8 +72,8 @@ def test_workload_plan_defaults_the_termination_grace_period(): plan = _plan() plan.validate_and_default() - assert DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS == 15 - assert plan.termination_grace_period_seconds == 15 + assert DEFAULT_TERMINATION_GRACE_PERIOD_SECONDS == 30 + assert plan.termination_grace_period_seconds == 30 def test_workload_plan_keeps_a_zero_termination_grace_period(): @@ -87,7 +89,7 @@ def test_workload_plan_defaults_a_none_termination_grace_period(): plan = _plan(termination_grace_period_seconds=None) plan.validate_and_default() - assert plan.termination_grace_period_seconds == 15 + assert plan.termination_grace_period_seconds == 30 def test_workload_plan_rejects_a_negative_termination_grace_period(): @@ -131,9 +133,12 @@ def __init__( self.name = name self.labels = {**(labels or {}), component_label: component} self.journal = journal + self.stop_error = None def stop(self, **kwargs): 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)) @@ -177,6 +182,25 @@ def _deployer(containers: list) -> SimpleNamespace: ) +_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 @@ -186,29 +210,39 @@ def test_docker_delete_drains_the_workload_before_removing_it(): DockerDeployer._delete(dep, name="test", grace_period_seconds=20) - assert journal == [ - ("remove", "test-unhealthy-restart", {"force": True}), + 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}), - ("remove", "test-init-0", {"force": True}), ("stop", "test-run-1", {"timeout": 20}), - ("remove", "test-run-1", {"force": True}), - ("remove", "test-pause", {"force": True}), ] +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: "30"}), + _docker_containers(journal, labels={_DOCKER_LABEL_GRACE_PERIOD: "45"}), ) DockerDeployer._delete(dep, name="test") - assert [c for c in journal if c[0] == "stop"] == [ - ("stop", "test-init-0", {"timeout": 30}), - ("stop", "test-run-1", {"timeout": 30}), + assert _stops(journal) == [ + ("stop", "test-init-0", {"timeout": 45}), + ("stop", "test-run-1", {"timeout": 45}), ] @@ -219,26 +253,40 @@ def test_docker_delete_falls_back_to_the_default_grace_period(): DockerDeployer._delete(dep, name="test") - assert [c for c in journal if c[0] == "stop"] == [ - ("stop", "test-init-0", {"timeout": 15}), - ("stop", "test-run-1", {"timeout": 15}), + 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: "30"}), + _docker_containers(journal, labels={_DOCKER_LABEL_GRACE_PERIOD: "45"}), ) DockerDeployer._delete(dep, name="test", grace_period_seconds=0) - assert [c for c in journal if c[0] == "stop"] == [ + assert _stops(journal) == [ ("stop", "test-init-0", {"timeout": 0}), ("stop", "test-run-1", {"timeout": 0}), ] +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. @@ -272,7 +320,7 @@ def test_docker_workload_plan_stamps_the_defaulted_grace_period_label(): ) plan.validate_and_default() - assert plan.labels[_DOCKER_LABEL_GRACE_PERIOD] == "15" + assert plan.labels[_DOCKER_LABEL_GRACE_PERIOD] == "30" def test_podman_delete_drains_the_workload_before_removing_it(): @@ -283,27 +331,26 @@ def test_podman_delete_drains_the_workload_before_removing_it(): PodmanDeployer._delete(dep, name="test", grace_period_seconds=20) - assert journal == [ - ("remove", "test-unhealthy-restart", {"force": True}), + 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}), - ("remove", "test-init-0", {"force": True}), ("stop", "test-run-1", {"timeout": 20, "ignore": True}), - ("remove", "test-run-1", {"force": True}), - ("remove", "test-pause", {"force": True}), ] def test_podman_delete_reads_the_grace_period_from_the_container_label(): journal = [] dep = _deployer( - _podman_containers(journal, labels={_PODMAN_LABEL_GRACE_PERIOD: "30"}), + _podman_containers(journal, labels={_PODMAN_LABEL_GRACE_PERIOD: "45"}), ) PodmanDeployer._delete(dep, name="test") - assert [c for c in journal if c[0] == "stop"] == [ - ("stop", "test-init-0", {"timeout": 30, "ignore": True}), - ("stop", "test-run-1", {"timeout": 30, "ignore": True}), + assert _stops(journal) == [ + ("stop", "test-init-0", {"timeout": 45, "ignore": True}), + ("stop", "test-run-1", {"timeout": 45, "ignore": True}), ] @@ -313,21 +360,21 @@ def test_podman_delete_falls_back_to_the_default_grace_period(): PodmanDeployer._delete(dep, name="test") - assert [c for c in journal if c[0] == "stop"] == [ - ("stop", "test-init-0", {"timeout": 15, "ignore": True}), - ("stop", "test-run-1", {"timeout": 15, "ignore": True}), + 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: "30"}), + _podman_containers(journal, labels={_PODMAN_LABEL_GRACE_PERIOD: "45"}), ) PodmanDeployer._delete(dep, name="test", grace_period_seconds=0) - assert [c for c in journal if c[0] == "stop"] == [ + assert _stops(journal) == [ ("stop", "test-init-0", {"timeout": 0, "ignore": True}), ("stop", "test-run-1", {"timeout": 0, "ignore": True}), ] @@ -492,3 +539,57 @@ def test_kubernetes_delete_forwards_the_grace_period_on_the_fallback_path(monkey 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) From f5b2cc65338256ff9c47508aea6e3c6ac0e54afd Mon Sep 17 00:00:00 2001 From: thxCode Date: Wed, 19 Aug 2026 12:41:53 +0800 Subject: [PATCH 7/7] fix(deployer): address the ship-time review - Compare Pod resource quantities by the number they denote, instead of rewriting them on the way out to match what the API server stores, which was both lossy and blind to the suffixed spellings the API server rewrites too. - Refuse a quantity Kubernetes would not accept, e.g. an underscore separated or a full-width one, so an invalid declaration cannot pass as an unchanged one and skip the deployment silently. - Report every Kubernetes workload deletion failure as an operation error as well, keeping the API error arm the 405 fallback reads the status from. - Accept the overriding grace period as a keyword argument only. - Bound the concurrent drain to 16 workers. - Pin the concurrent drain with a barrier, on both Docker and Podman, as the ordering assertion alone still passed against a drain that signalled the containers one at a time. Review follow-ups of graceful-workload-termination. Signed-off-by: thxCode --- gpustack_runtime/deployer/__types__.py | 1 + gpustack_runtime/deployer/docker.py | 2 +- gpustack_runtime/deployer/kuberentes.py | 105 +++++++++++++- gpustack_runtime/deployer/podman.py | 2 +- .../deployer/test_termination_grace_period.py | 133 ++++++++++++++++++ 5 files changed, 234 insertions(+), 9 deletions(-) diff --git a/gpustack_runtime/deployer/__types__.py b/gpustack_runtime/deployer/__types__.py index d15c215..53cb6ea 100644 --- a/gpustack_runtime/deployer/__types__.py +++ b/gpustack_runtime/deployer/__types__.py @@ -2132,6 +2132,7 @@ def delete( self, name: WorkloadName, namespace: WorkloadNamespace | None = None, + *, grace_period_seconds: int | None = None, async_mode: bool | None = None, ) -> WorkloadStatus | None: diff --git a/gpustack_runtime/deployer/docker.py b/gpustack_runtime/deployer/docker.py index 50a5cae..e37b21d 100644 --- a/gpustack_runtime/deployer/docker.py +++ b/gpustack_runtime/deployer/docker.py @@ -2236,7 +2236,7 @@ def stop(container: docker.models.containers.Container): except Exception: debug_log_exception(logger, f"Failed to stop container {container.name}") - with ThreadPoolExecutor(max_workers=len(containers)) as pool: + with ThreadPoolExecutor(max_workers=min(len(containers), 16)) as pool: list(pool.map(stop, containers)) diff --git a/gpustack_runtime/deployer/kuberentes.py b/gpustack_runtime/deployer/kuberentes.py index 0a301e1..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 @@ -2268,9 +2270,14 @@ def _delete( 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: @@ -2297,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: @@ -2324,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 @@ -2853,18 +2870,89 @@ 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, + 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 {k: v for k, v in container.resources.to_dict().items() if v} + 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( @@ -3049,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. @@ -3065,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 27306d6..85cfdfa 100644 --- a/gpustack_runtime/deployer/podman.py +++ b/gpustack_runtime/deployer/podman.py @@ -2183,7 +2183,7 @@ def stop(container: podman.domain.containers.Container): except Exception: debug_log_exception(logger, f"Failed to stop container {container.name}") - with ThreadPoolExecutor(max_workers=len(containers)) as pool: + with ThreadPoolExecutor(max_workers=min(len(containers), 16)) as pool: list(pool.map(stop, containers)) diff --git a/tests/gpustack_runtime/deployer/test_termination_grace_period.py b/tests/gpustack_runtime/deployer/test_termination_grace_period.py index c02fd75..f2e9323 100644 --- a/tests/gpustack_runtime/deployer/test_termination_grace_period.py +++ b/tests/gpustack_runtime/deployer/test_termination_grace_period.py @@ -3,11 +3,13 @@ # 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 ( @@ -15,6 +17,7 @@ Container, ContainerProfileEnum, Deployer, + OperationError, WorkloadPlan, ) from gpustack_runtime.deployer.docker import ( @@ -30,6 +33,7 @@ from gpustack_runtime.deployer.kuberentes import ( KubernetesDeployer, KubernetesWorkloadPlan, + equal_containers, equal_pods, ) from gpustack_runtime.deployer.podman import ( @@ -134,8 +138,13 @@ def __init__( 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 @@ -273,6 +282,42 @@ def test_docker_delete_kills_immediately_on_a_zero_grace_period(): ] +@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. @@ -397,6 +442,47 @@ def test_podman_workload_plan_stamps_the_grace_period_label(): 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. @@ -593,3 +679,50 @@ def test_kubernetes_equal_pods_detects_a_changed_grace_period(): 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))