From 33be87bf14f173a7f4d0ddec0c257ff44812e180 Mon Sep 17 00:00:00 2001 From: Victor Skvortsov Date: Tue, 3 Nov 2026 12:42:20 +0500 Subject: [PATCH 1/2] Support orphan cleanup in the Kubernetes backend --- mkdocs/docs/reference/env.md | 2 +- .../kubernetes/dstack-backend-role.yaml | 4 +- .../core/backends/kubernetes/compute.py | 77 +++--- .../backends/kubernetes/features/__init__.py | 0 .../kubernetes/features/resource_groups.py | 211 ++++++++++++++ .../core/backends/kubernetes/resources.py | 47 +++- .../core/backends/kubernetes/utils.py | 38 ++- .../backends/kubernetes/features/__init__.py | 0 .../features/test_resource_groups.py | 259 ++++++++++++++++++ .../core/backends/kubernetes/test_utils.py | 33 +++ 10 files changed, 622 insertions(+), 49 deletions(-) create mode 100644 src/dstack/_internal/core/backends/kubernetes/features/__init__.py create mode 100644 src/dstack/_internal/core/backends/kubernetes/features/resource_groups.py create mode 100644 src/tests/_internal/core/backends/kubernetes/features/__init__.py create mode 100644 src/tests/_internal/core/backends/kubernetes/features/test_resource_groups.py diff --git a/mkdocs/docs/reference/env.md b/mkdocs/docs/reference/env.md index 810645974f..3c3b258b2c 100644 --- a/mkdocs/docs/reference/env.md +++ b/mkdocs/docs/reference/env.md @@ -153,7 +153,7 @@ For more details on the options below, refer to the [server deployment](../guide - `DSTACK_SERVER_GATEWAY_MAX_REPLICAS`{ #DSTACK_SERVER_GATEWAY_MAX_REPLICAS } - Maximum number of replicas allowed for a gateway. Defaults to `9`. - `DSTACK_SERVER_INSTANCE_HEALTH_MIN_COLLECT_INTERVAL_SECONDS`{ #DSTACK_SERVER_INSTANCE_HEALTH_MIN_COLLECT_INTERVAL_SECONDS } – Minimum time interval between consecutive health checks of the same instance. - `DSTACK_SERVER_EVENTS_TTL_SECONDS`{ #DSTACK_SERVER_EVENTS_TTL_SECONDS } - Maximum age of event records. Set to `0` to disable event storage. Defaults to 30 days. -- `DSTACK_SERVER_ORPHAN_CLEANUP_MODE`{ #DSTACK_SERVER_ORPHAN_CLEANUP_MODE } – Controls the cleanup of cloud resources that `dstack` created but no longer tracks, e.g. after the server crashed during provisioning or failed to terminate an instance. `report` emits events about such resources, `delete` also deletes them, and `off` disables the cleanup. Resources are recognized by the tags (or names) `dstack` sets on them, so a resource created by copying the tags of a `dstack` instance is treated as an orphan and deleted in `delete` mode. Defaults to `report`. Currently supported for the `aws`, `azure`, `gcp`, `nebius`, and `runpod` backends. +- `DSTACK_SERVER_ORPHAN_CLEANUP_MODE`{ #DSTACK_SERVER_ORPHAN_CLEANUP_MODE } – Controls the cleanup of cloud resources that `dstack` created but no longer tracks, e.g. after the server crashed during provisioning or failed to terminate an instance. `report` emits events about such resources, `delete` also deletes them, and `off` disables the cleanup. Resources are recognized by the tags (or names) `dstack` sets on them, so a resource created by copying the tags of a `dstack` instance is treated as an orphan and deleted in `delete` mode. Defaults to `report`. Currently supported for the `aws`, `azure`, `gcp`, `kubernetes`, `nebius`, and `runpod` backends. - `DSTACK_SERVER_ORPHAN_CLEANUP_INTERVAL_SECONDS`{ #DSTACK_SERVER_ORPHAN_CLEANUP_INTERVAL_SECONDS } – Minimum interval between orphan cleanups of the same backend. A backend is cleaned up only after instances were provisioned or terminated in its project. Defaults to `3600`. - `DSTACK_SERVER_DEFAULT_DOCKER_REGISTRY`{ #DSTACK_SERVER_DEFAULT_DOCKER_REGISTRY } – A default Docker registry to use for job images that do not specify an explicit registry. E.g., if set to `registry.example`, then `image: ubuntu` becomes equivalent to `image: registry.example/ubuntu`. **Note**: This setting should only be used for configuring registries that act as a pull-through cache for Docker Hub. The default `dstack` images are also pulled from the configured registry. - `DSTACK_SERVER_DEFAULT_DOCKER_REGISTRY_USERNAME`{ #DSTACK_SERVER_DEFAULT_DOCKER_REGISTRY_USERNAME } – Username for authenticating with the default Docker registry. See `DSTACK_SERVER_DEFAULT_DOCKER_REGISTRY_PASSWORD`. diff --git a/mkdocs/snippets/kubernetes/dstack-backend-role.yaml b/mkdocs/snippets/kubernetes/dstack-backend-role.yaml index c2c9ca5145..deddedfe3c 100644 --- a/mkdocs/snippets/kubernetes/dstack-backend-role.yaml +++ b/mkdocs/snippets/kubernetes/dstack-backend-role.yaml @@ -9,10 +9,10 @@ rules: verbs: ["get", "list", "watch", "create", "delete"] - apiGroups: [""] resources: ["services"] - verbs: ["get", "create", "delete"] + verbs: ["get", "list", "create", "delete"] - apiGroups: [""] resources: ["secrets"] - verbs: ["create", "delete"] + verbs: ["list", "create", "delete"] - apiGroups: [""] resources: ["persistentvolumeclaims"] verbs: ["get", "create", "delete"] diff --git a/src/dstack/_internal/core/backends/kubernetes/compute.py b/src/dstack/_internal/core/backends/kubernetes/compute.py index c9cd0e7f9f..0ab3beda53 100644 --- a/src/dstack/_internal/core/backends/kubernetes/compute.py +++ b/src/dstack/_internal/core/backends/kubernetes/compute.py @@ -26,6 +26,7 @@ ComputeWithInstanceVolumesSupport, ComputeWithMultinodeSupport, ComputeWithPrivilegedSupport, + ComputeWithResourceListingSupport, ComputeWithVolumeSupport, generate_unique_gateway_instance_name, generate_unique_instance_name_for_job, @@ -33,6 +34,7 @@ generate_unique_volume_name, get_docker_commands, get_dstack_gateway_install_command, + get_job_provider_ref, merge_tags, ) from dstack._internal.core.backends.base.offers import ( @@ -41,11 +43,13 @@ gpu_matches_gpu_spec, ) from dstack._internal.core.backends.kubernetes.api_client import API_CLIENT_EXCEPTIONS +from dstack._internal.core.backends.kubernetes.features import resource_groups from dstack._internal.core.backends.kubernetes.models import KubernetesConfig from dstack._internal.core.backends.kubernetes.resources import ( AMD_GPU_DEVICE_ID_LABEL_PREFIX, AMD_GPU_NAME_TO_DEVICE_IDS, AMD_GPU_NODE_TAINT, + INSTANCE_LABEL, LABEL_VALUE_MAX_LENGTH, NVIDIA_GPU_NODE_TAINT, NVIDIA_GPU_PRODUCT_LABEL, @@ -59,6 +63,7 @@ adjust_resources_by_resource_requests, build_base_labels, build_dockerconfigjson, + delete_job_pod_objects, filter_invalid_labels, format_memory, get_amd_gpu_from_node_labels, @@ -67,6 +72,8 @@ get_nvidia_gpu_from_node_labels, get_object_labels, get_object_name, + get_pod_service_name, + get_registry_auth_secret_name, is_hard_taint, is_taint_tolerated, parse_quantity, @@ -82,6 +89,10 @@ from dstack._internal.core.consts import DSTACK_RUNNER_SSH_PORT from dstack._internal.core.errors import ComputeError, ProvisioningError, SkipOffer from dstack._internal.core.models.backends.base import BackendType +from dstack._internal.core.models.backends.orphans import ( + ListedResourceGroup, + ResourceListing, +) from dstack._internal.core.models.common import CoreModel, validate_json_extra_ignore from dstack._internal.core.models.gateways import ( GatewayReplicaConfiguration, @@ -161,6 +172,7 @@ class KubernetesCompute( ComputeWithVolumeSupport, ComputeWithGatewaySupport, ComputeWithMultinodeSupport, + ComputeWithResourceListingSupport, Compute, ): unallocated_resources_argument_has_effect = True @@ -235,7 +247,7 @@ def run_job( # to all job pods of the same project. # The service is created here and configured later in update_provisioning_data() jump_pod_name = f"dstack-{run.project_name}-ssh-jump-pod" - jump_pod_service_name = _get_pod_service_name(jump_pod_name) + jump_pod_service_name = get_pod_service_name(jump_pod_name) _create_jump_pod_service_if_not_exists( api=api, apps_api=client.AppsV1Api(cluster.api_client), @@ -257,6 +269,7 @@ def run_job( project=run.project_name, name=job.job_spec.job_name, user=run.user, + provider_ref=get_job_provider_ref(job), ) labels = merge_tags( base_tags=base_labels, @@ -267,7 +280,7 @@ def run_job( registry_auth_secret_name: Optional[str] = None with ExitStack() as exit_stack: if job.job_spec.registry_auth is not None: - registry_auth_secret_name = _get_registry_auth_secret_name(pod_name) + registry_auth_secret_name = get_registry_auth_secret_name(pod_name) _create_registry_auth_secret( api=api, namespace=namespace, @@ -327,7 +340,7 @@ def run_job( # It's not clear if we should add an entry to the SkipOfferCache in this case. raise ComputeError(f"Pod {pod_name} already finished: {pod_phase}") - pod_service_name = _get_pod_service_name(pod_name) + pod_service_name = get_pod_service_name(pod_name) api.create_namespaced_service( namespace=namespace, body=client.V1Service( @@ -432,7 +445,7 @@ def update_provisioning_data( return provisioning_data.internal_ip = pod_ip service = api.read_namespaced_service( - name=_get_pod_service_name(provisioning_data.instance_id), + name=get_pod_service_name(provisioning_data.instance_id), namespace=namespace, ) service_spec = get_or_error(service.spec) @@ -475,30 +488,17 @@ def terminate_instance( ) if cluster is None: raise ComputeError(f"Unknown region: {region!r}") - api = client.CoreV1Api(cluster.api_client) - namespace = cluster.namespace - deleted = [ - try_delete_object_if_exists( - api.delete_namespaced_service, - namespace=namespace, - name=_get_pod_service_name(instance_id), - description="pod service", - ), - try_delete_object_if_exists( - api.delete_namespaced_pod, - namespace=namespace, - name=instance_id, - description="pod", - ), - try_delete_object_if_exists( - api.delete_namespaced_secret, - namespace=namespace, - name=_get_registry_auth_secret_name(instance_id), - description="registry auth secret", - ), - ] - if not all(deleted): - raise ComputeError("Not all objects were deleted, check logs") + delete_job_pod_objects( + api=client.CoreV1Api(cluster.api_client), + namespace=cluster.namespace, + pod_name=instance_id, + ) + + def list_resource_groups(self) -> ResourceListing: + return resource_groups.list_resource_groups(list(self.region_cluster_map.values())) + + def delete_resource_group(self, group: ListedResourceGroup) -> None: + resource_groups.delete_resource_group(self.region_cluster_map, group) def create_gateway_replica( self, @@ -579,7 +579,7 @@ def create_gateway_replica( ) service = client.V1Service( metadata=client.V1ObjectMeta( - name=_get_pod_service_name(instance_name), + name=get_pod_service_name(instance_name), labels=labels, ), spec=client.V1ServiceSpec( @@ -612,7 +612,7 @@ def create_gateway_replica( address = _wait_for_load_balancer_address( api=api, namespace=namespace, - service_name=_get_pod_service_name(instance_name), + service_name=get_pod_service_name(instance_name), ) if address is None: self.terminate_instance(instance_name, region=configuration.region) @@ -1569,25 +1569,14 @@ def _run_ssh_command( return proc.returncode, proc.stdout -_SELECTOR_LABEL_KEY = "app.kubernetes.io/instance" - - def _build_service_selector_from_labels(labels: dict[str, str]) -> dict[str, str]: - return {_SELECTOR_LABEL_KEY: labels[_SELECTOR_LABEL_KEY]} + return {INSTANCE_LABEL: labels[INSTANCE_LABEL]} def _build_label_selector_from_labels(labels: dict[str, str]) -> client.V1LabelSelector: - return client.V1LabelSelector(match_labels={_SELECTOR_LABEL_KEY: labels[_SELECTOR_LABEL_KEY]}) + return client.V1LabelSelector(match_labels={INSTANCE_LABEL: labels[INSTANCE_LABEL]}) def _build_query_label_selector_from_unique_name(unique_name: str) -> str: # See build_base_labels() - return f"{_SELECTOR_LABEL_KEY}={unique_name}" - - -def _get_pod_service_name(pod_name: str) -> str: - return f"{pod_name}-service" - - -def _get_registry_auth_secret_name(pod_name: str) -> str: - return f"{pod_name}-registry-auth" + return f"{INSTANCE_LABEL}={unique_name}" diff --git a/src/dstack/_internal/core/backends/kubernetes/features/__init__.py b/src/dstack/_internal/core/backends/kubernetes/features/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/src/dstack/_internal/core/backends/kubernetes/features/resource_groups.py b/src/dstack/_internal/core/backends/kubernetes/features/resource_groups.py new file mode 100644 index 0000000000..15a2ec25ca --- /dev/null +++ b/src/dstack/_internal/core/backends/kubernetes/features/resource_groups.py @@ -0,0 +1,211 @@ +from dataclasses import dataclass, field +from datetime import datetime +from typing import Callable, Optional, TypeVar + +from kubernetes import client +from kubernetes.client.exceptions import ApiException + +from dstack._internal.core.backends.base.compute import is_valid_provider_ref +from dstack._internal.core.backends.kubernetes.api_client import API_CLIENT_EXCEPTIONS +from dstack._internal.core.backends.kubernetes.resources import ( + INSTANCE_LABEL, + PROVIDER_REF_LABEL, + ObjectWithMeta, + delete_job_pod_objects, + get_object_labels, + get_object_name, +) +from dstack._internal.core.backends.kubernetes.utils import ( + Cluster, + ObjectListWithMeta, + list_all_objects, +) +from dstack._internal.core.errors import ComputeError +from dstack._internal.core.models.backends.base import BackendType +from dstack._internal.core.models.backends.orphans import ( + ListedResourceGroup, + OwnerType, + ResourceListing, + ResourceListingFailure, +) +from dstack._internal.core.models.common import CoreModel, validate_json_extra_ignore +from dstack._internal.utils.common import get_or_error +from dstack._internal.utils.logging import get_logger + +logger = get_logger(__name__) + + +class KubernetesResourceGroupBackendData(CoreModel): + pod_name: str + """The job pod's name. The names of its service and secret are derived from it.""" + + +def list_resource_groups(clusters: list[Cluster]) -> ResourceListing: + """ + Each group is a job pod with its service and registry auth secret, listed by the + `provider_ref` label in the namespace of each cluster. + """ + resource_groups: list[ListedResourceGroup] = [] + failures: list[ResourceListingFailure] = [] + for cluster in clusters: + try: + resource_groups.extend(_list_cluster_resource_groups(cluster)) + except (*API_CLIENT_EXCEPTIONS, ComputeError) as e: + failures.append( + ResourceListingFailure( + backend=BackendType.KUBERNETES, + region=cluster.region, + message=f"Failed to list objects in cluster {cluster}: {e}", + ) + ) + if len(failures) > 0 and len(failures) == len(clusters): + raise ComputeError(f"Failed to list resources in all clusters: {failures}") + return ResourceListing(resource_groups=resource_groups, failures=failures) + + +def delete_resource_group( + clusters_by_region: dict[str, Cluster], group: ListedResourceGroup +) -> None: + cluster = clusters_by_region.get(get_or_error(group.region)) + if cluster is None: + raise ComputeError(f"Unknown region: {group.region!r}") + backend_data = validate_json_extra_ignore( + KubernetesResourceGroupBackendData, get_or_error(group.backend_data) + ) + delete_job_pod_objects( + api=client.CoreV1Api(cluster.api_client), + namespace=cluster.namespace, + pod_name=backend_data.pod_name, + ) + + +_T = TypeVar("_T") + + +@dataclass +class _JobPodObjects: + provider_ref: str + pod_name: str + pod: Optional[client.V1Pod] = None + internal_object_descriptions: list[str] = field(default_factory=list) + created_at: Optional[datetime] = None + + def add_created_at(self, created_at: Optional[datetime]) -> None: + if created_at is not None and (self.created_at is None or created_at < self.created_at): + self.created_at = created_at + + +def _list_cluster_resource_groups(cluster: Cluster) -> list[ListedResourceGroup]: + """ + Pods must be listed. Services and secrets are skipped with a warning if listing them is + forbidden: groups with pods still delete them by their derived names, and only those left + without pods are not found. + """ + api = client.CoreV1Api(cluster.api_client) + namespace = cluster.namespace + pods = list_all_objects( + api.list_namespaced_pod, namespace=namespace, label_selector=PROVIDER_REF_LABEL + ) + services = _list_objects_with_provider_ref_if_permitted( + api.list_namespaced_service, namespace, cluster=cluster, kind="services" + ) + secrets = _list_objects_with_provider_ref_if_permitted( + api.list_namespaced_secret, namespace, cluster=cluster, kind="secrets" + ) + groups: dict[tuple[str, str], _JobPodObjects] = {} + + def get_group(obj: ObjectWithMeta) -> Optional[_JobPodObjects]: + key = _get_job_pod_objects_key(obj, cluster) + if key is None: + return None + if key not in groups: + groups[key] = _JobPodObjects(provider_ref=key[0], pod_name=key[1]) + return groups[key] + + for pod in pods: + metadata = get_or_error(pod.metadata) + # The pod is being deleted; its service and secret are listed as usual. + if metadata.deletion_timestamp is not None: + continue + group = get_group(pod) + if group is None: + continue + group.pod = pod + group.add_created_at(metadata.creation_timestamp) + for kind, objects in (("service", services), ("secret", secrets)): + for obj in objects: + group = get_group(obj) + if group is None: + continue + metadata = get_or_error(obj.metadata) + group.internal_object_descriptions.append(f"{kind} {metadata.name}") + if group.pod is None: + group.add_created_at(metadata.creation_timestamp) + return [_build_resource_group(group, cluster) for group in groups.values()] + + +def _list_objects_with_provider_ref_if_permitted( + method: Callable[..., ObjectListWithMeta[_T]], namespace: str, *, cluster: Cluster, kind: str +) -> list[_T]: + try: + return list_all_objects(method, namespace=namespace, label_selector=PROVIDER_REF_LABEL) + except ApiException as e: + if e.status != 403: + raise + logger.warning( + "Cannot list %s in namespace %s of cluster %s for orphan cleanup." + " Grant the `list` verb on %s to clean up %s left without their pods", + kind, + namespace, + cluster, + kind, + kind, + ) + return [] + + +def _get_job_pod_objects_key(obj: ObjectWithMeta, cluster: Cluster) -> Optional[tuple[str, str]]: + labels = get_object_labels(obj) + object_name = get_object_name(obj) + provider_ref = labels.get(PROVIDER_REF_LABEL) + if provider_ref is None: + return None + if not is_valid_provider_ref(provider_ref): + logger.warning( + "Skipping object %s in cluster %s with invalid %s label: %r", + object_name, + cluster, + PROVIDER_REF_LABEL, + provider_ref, + ) + return None + pod_name = labels.get(INSTANCE_LABEL) + if pod_name is None: + logger.warning( + "Skipping object %s in cluster %s without %s label", + object_name, + cluster, + INSTANCE_LABEL, + ) + return None + return provider_ref, pod_name + + +def _build_resource_group(group: _JobPodObjects, cluster: Cluster) -> ListedResourceGroup: + description_parts: list[str] = [] + primary_id: Optional[str] = None + if group.pod is not None: + primary_id = group.pod_name + phase = group.pod.status.phase if group.pod.status is not None else None + description_parts.append(f"pod {group.pod_name} ({phase or 'unknown phase'})") + description_parts.extend(group.internal_object_descriptions) + return ListedResourceGroup( + backend=BackendType.KUBERNETES, + owner_type=OwnerType.INSTANCE, + provider_ref=group.provider_ref, + primary_id=primary_id, + region=cluster.region, + created_at=group.created_at, + description=", ".join(description_parts), + backend_data=KubernetesResourceGroupBackendData(pod_name=group.pod_name).model_dump_json(), + ) diff --git a/src/dstack/_internal/core/backends/kubernetes/resources.py b/src/dstack/_internal/core/backends/kubernetes/resources.py index 37aa36ef6b..3993154c57 100644 --- a/src/dstack/_internal/core/backends/kubernetes/resources.py +++ b/src/dstack/_internal/core/backends/kubernetes/resources.py @@ -14,6 +14,8 @@ from typing_extensions import Self from dstack._internal.core.backends.base.compute import normalize_arch +from dstack._internal.core.backends.kubernetes.utils import try_delete_object_if_exists +from dstack._internal.core.errors import ComputeError from dstack._internal.core.models.backends.base import BackendType from dstack._internal.core.models.instances import ( Disk, @@ -261,6 +263,13 @@ def adjust_resources_by_resource_requests( resources.disk = Disk(size_mib=disk_mib) +INSTANCE_LABEL = "app.kubernetes.io/instance" +"""The label carrying the object's unique name, shared by the objects of a job pod.""" + +PROVIDER_REF_LABEL = "k8s.dstack.ai/provider-ref" +"""The label carrying `provider_ref` for orphan cleanup.""" + + def build_base_labels( *, component: Literal["ssh-proxy", "job", "gateway", "volume"], @@ -268,12 +277,13 @@ def build_base_labels( project: str, name: Optional[str] = None, user: Optional[str] = None, + provider_ref: Optional[str] = None, ) -> dict[str, str]: labels = { "app.kubernetes.io/name": f"dstack-{component}", # app.kubernetes.io/component would be redundant as app.kubernetes.io/name already includes # it with dstack- prefix - "app.kubernetes.io/instance": unique_name, + INSTANCE_LABEL: unique_name, "app.kubernetes.io/managed-by": "dstack", "k8s.dstack.ai/project": project, } @@ -281,9 +291,44 @@ def build_base_labels( labels["k8s.dstack.ai/name"] = name if user is not None: labels["k8s.dstack.ai/user"] = user + if provider_ref is not None: + labels[PROVIDER_REF_LABEL] = provider_ref return labels +def get_pod_service_name(pod_name: str) -> str: + return f"{pod_name}-service" + + +def get_registry_auth_secret_name(pod_name: str) -> str: + return f"{pod_name}-registry-auth" + + +def delete_job_pod_objects(api: CoreV1Api, namespace: str, pod_name: str) -> None: + deleted = [ + try_delete_object_if_exists( + api.delete_namespaced_service, + namespace=namespace, + name=get_pod_service_name(pod_name), + description="pod service", + ), + try_delete_object_if_exists( + api.delete_namespaced_pod, + namespace=namespace, + name=pod_name, + description="pod", + ), + try_delete_object_if_exists( + api.delete_namespaced_secret, + namespace=namespace, + name=get_registry_auth_secret_name(pod_name), + description="registry auth secret", + ), + ] + if not all(deleted): + raise ComputeError("Not all objects were deleted, check logs") + + def filter_invalid_labels(labels: dict[str, str]) -> dict[str, str]: filtered_labels: dict[str, str] = {} for k, v in labels.items(): diff --git a/src/dstack/_internal/core/backends/kubernetes/utils.py b/src/dstack/_internal/core/backends/kubernetes/utils.py index 7907df7341..18d58eec00 100644 --- a/src/dstack/_internal/core/backends/kubernetes/utils.py +++ b/src/dstack/_internal/core/backends/kubernetes/utils.py @@ -15,7 +15,7 @@ ) import yaml -from kubernetes.client import V1Status, VersionApi +from kubernetes.client import V1ListMeta, V1Status, VersionApi from kubernetes.client.exceptions import ApiException from kubernetes.watch import Watch from pydantic import Field @@ -30,6 +30,7 @@ KubernetesBackendConfigWithCreds, KubernetesProxyJumpConfig, ) +from dstack._internal.core.errors import ComputeError from dstack._internal.core.models.common import CoreModel, validate_extra_ignore from dstack._internal.utils.logging import get_logger @@ -268,6 +269,41 @@ class ObjectList(Protocol[T]): items: list[T] +class ObjectListWithMeta(ObjectList[T], Protocol[T]): + metadata: Optional[V1ListMeta] + + +_LIST_PAGE_SIZE = 500 +_LIST_MAX_PAGES = 1000 +"""Guards against an API that never stops returning continue tokens.""" + + +def list_all_objects( + method: Callable[..., ObjectListWithMeta[T]], *, namespace: str, label_selector: str +) -> list[T]: + """ + Lists all objects of a namespaced kind matching `label_selector`, page by page. + Raises `ComputeError` instead of returning a partial list if there are too many pages. + """ + objects: list[T] = [] + continue_token: Optional[str] = None + for _ in range(_LIST_MAX_PAGES): + object_list = method( + namespace=namespace, + label_selector=label_selector, + limit=_LIST_PAGE_SIZE, + _continue=continue_token, + ) + objects.extend(object_list.items) + continue_token = object_list.metadata._continue if object_list.metadata else None + if not continue_token: + return objects + raise ComputeError( + f"Listing with {getattr(method, '__name__', method)} in namespace {namespace}" + f" exceeded the limit of {_LIST_MAX_PAGES} pages" + ) + + @contextmanager def watch_events( method: Callable[P, ObjectList[T]], *args: P.args, **kwargs: P.kwargs diff --git a/src/tests/_internal/core/backends/kubernetes/features/__init__.py b/src/tests/_internal/core/backends/kubernetes/features/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/src/tests/_internal/core/backends/kubernetes/features/test_resource_groups.py b/src/tests/_internal/core/backends/kubernetes/features/test_resource_groups.py new file mode 100644 index 0000000000..1aaeb451ee --- /dev/null +++ b/src/tests/_internal/core/backends/kubernetes/features/test_resource_groups.py @@ -0,0 +1,259 @@ +from collections.abc import Generator +from datetime import datetime, timezone +from typing import Optional +from unittest.mock import Mock, patch + +import pytest +from kubernetes import client +from kubernetes.client.exceptions import ApiException + +from dstack._internal.core.backends.kubernetes.features.resource_groups import ( + delete_resource_group, + list_resource_groups, +) +from dstack._internal.core.backends.kubernetes.models import KubernetesProxyJumpConfig +from dstack._internal.core.backends.kubernetes.resources import build_base_labels +from dstack._internal.core.backends.kubernetes.utils import Cluster +from dstack._internal.core.errors import ComputeError +from dstack._internal.core.models.backends.base import BackendType +from dstack._internal.core.models.backends.orphans import ListedResourceGroup, OwnerType + +NAMESPACE = "dstack" +PROVIDER_REF = "abcdefgh1234" +JOB_POD_NAME = "dstack-run-0-0-abcd" +CREATED_AT = datetime(2026, 1, 1, tzinfo=timezone.utc) + + +def _job_labels(pod_name: str = JOB_POD_NAME, provider_ref: str = PROVIDER_REF) -> dict[str, str]: + return build_base_labels( + component="job", unique_name=pod_name, project="main", provider_ref=provider_ref + ) + + +def _job_object_meta( + name: str, + *, + labels: Optional[dict[str, str]] = None, + created_at: datetime = CREATED_AT, + terminating: bool = False, +) -> client.V1ObjectMeta: + return client.V1ObjectMeta( + name=name, + labels=labels if labels is not None else _job_labels(), + creation_timestamp=created_at, + deletion_timestamp=CREATED_AT if terminating else None, + ) + + +def _object_list(items: list, continue_token: Optional[str] = None) -> Mock: + return Mock(items=items, metadata=client.V1ListMeta(_continue=continue_token)) + + +def _cluster(region: str = "ctx") -> Cluster: + return Cluster( + context_name=region, + region=region, + api_client=Mock(name=f"api_client_{region}"), + namespace=NAMESPACE, + proxy_jump=KubernetesProxyJumpConfig(), + ) + + +class TestListResourceGroups: + @pytest.fixture + def api(self) -> Generator[Mock, None, None]: + api = Mock() + api.list_namespaced_pod.return_value = _object_list([]) + api.list_namespaced_service.return_value = _object_list([]) + api.list_namespaced_secret.return_value = _object_list([]) + with patch( + "dstack._internal.core.backends.kubernetes.features.resource_groups.client.CoreV1Api", + return_value=api, + ): + yield api + + def test_groups_pod_with_its_service_and_secret(self, api: Mock): + api.list_namespaced_pod.return_value = _object_list( + [ + client.V1Pod( + metadata=_job_object_meta(JOB_POD_NAME), + status=client.V1PodStatus(phase="Running"), + ) + ] + ) + api.list_namespaced_service.return_value = _object_list( + [client.V1Service(metadata=_job_object_meta(f"{JOB_POD_NAME}-service"))] + ) + api.list_namespaced_secret.return_value = _object_list( + [client.V1Secret(metadata=_job_object_meta(f"{JOB_POD_NAME}-registry-auth"))] + ) + + listing = list_resource_groups([_cluster()]) + + assert listing.failures == [] + assert len(listing.resource_groups) == 1 + group = listing.resource_groups[0] + assert group.provider_ref == PROVIDER_REF + assert group.primary_id == JOB_POD_NAME + assert group.region == "ctx" + assert group.created_at == CREATED_AT + assert group.description == ( + f"pod {JOB_POD_NAME} (Running), service {JOB_POD_NAME}-service," + f" secret {JOB_POD_NAME}-registry-auth" + ) + api.list_namespaced_pod.assert_called_once_with( + namespace=NAMESPACE, + label_selector="k8s.dstack.ai/provider-ref", + limit=500, + _continue=None, + ) + + def test_lists_finished_pods(self, api: Mock): + api.list_namespaced_pod.return_value = _object_list( + [ + client.V1Pod( + metadata=_job_object_meta(JOB_POD_NAME), + status=client.V1PodStatus(phase="Succeeded"), + ) + ] + ) + + listing = list_resource_groups([_cluster()]) + + assert [g.primary_id for g in listing.resource_groups] == [JOB_POD_NAME] + + def test_lists_internal_objects_of_pod_being_deleted_without_pod(self, api: Mock): + api.list_namespaced_pod.return_value = _object_list( + [client.V1Pod(metadata=_job_object_meta(JOB_POD_NAME, terminating=True))] + ) + api.list_namespaced_service.return_value = _object_list( + [ + client.V1Service( + metadata=_job_object_meta( + f"{JOB_POD_NAME}-service", created_at=datetime(2026, 1, 2) + ) + ) + ] + ) + + listing = list_resource_groups([_cluster()]) + + assert len(listing.resource_groups) == 1 + group = listing.resource_groups[0] + assert group.primary_id is None + assert group.created_at == datetime(2026, 1, 2) + assert group.description == f"service {JOB_POD_NAME}-service" + + def test_skips_objects_without_valid_labels(self, api: Mock): + labels_without_instance = _job_labels() + del labels_without_instance["app.kubernetes.io/instance"] + api.list_namespaced_pod.return_value = _object_list( + [ + client.V1Pod(metadata=_job_object_meta("a", labels={})), + client.V1Pod(metadata=_job_object_meta("b", labels=_job_labels(provider_ref="x"))), + client.V1Pod(metadata=_job_object_meta("c", labels=labels_without_instance)), + ] + ) + + listing = list_resource_groups([_cluster()]) + + assert listing.resource_groups == [] + + def test_separates_attempts_by_pod(self, api: Mock): + api.list_namespaced_pod.return_value = _object_list( + [ + client.V1Pod(metadata=_job_object_meta(name, labels=_job_labels(pod_name=name))) + for name in ["pod-a", "pod-b"] + ] + ) + + listing = list_resource_groups([_cluster()]) + + assert sorted(g.primary_id or "" for g in listing.resource_groups) == ["pod-a", "pod-b"] + assert {g.provider_ref for g in listing.resource_groups} == {PROVIDER_REF} + + def test_skips_services_and_secrets_if_listing_them_is_forbidden(self, api: Mock): + api.list_namespaced_pod.return_value = _object_list( + [client.V1Pod(metadata=_job_object_meta(JOB_POD_NAME))] + ) + api.list_namespaced_service.side_effect = ApiException(status=403) + api.list_namespaced_secret.side_effect = ApiException(status=403) + + listing = list_resource_groups([_cluster()]) + + assert listing.failures == [] + assert [g.primary_id for g in listing.resource_groups] == [JOB_POD_NAME] + + def test_reports_cluster_failure(self, api: Mock): + pods = [client.V1Pod(metadata=_job_object_meta(JOB_POD_NAME))] + api.list_namespaced_pod.side_effect = [ApiException(status=403), _object_list(pods)] + + listing = list_resource_groups([_cluster("a"), _cluster("b")]) + + assert [f.region for f in listing.failures] == ["a"] + assert [g.region for g in listing.resource_groups] == ["b"] + + def test_fails_on_internal_object_listing_error_other_than_forbidden(self, api: Mock): + api.list_namespaced_pod.return_value = _object_list( + [client.V1Pod(metadata=_job_object_meta(JOB_POD_NAME))] + ) + api.list_namespaced_secret.side_effect = ApiException(status=500) + + with pytest.raises(ComputeError): + list_resource_groups([_cluster()]) + + def test_raises_if_all_clusters_fail(self, api: Mock): + api.list_namespaced_pod.side_effect = ApiException(status=500) + + with pytest.raises(ComputeError): + list_resource_groups([_cluster("a"), _cluster("b")]) + + +class TestDeleteResourceGroup: + @pytest.fixture + def api(self) -> Generator[Mock, None, None]: + api = Mock() + with patch( + "dstack._internal.core.backends.kubernetes.features.resource_groups.client.CoreV1Api", + return_value=api, + ): + yield api + + def group(self, primary_id: Optional[str] = JOB_POD_NAME) -> ListedResourceGroup: + return ListedResourceGroup( + backend=BackendType.KUBERNETES, + owner_type=OwnerType.INSTANCE, + provider_ref=PROVIDER_REF, + primary_id=primary_id, + region="ctx", + created_at=CREATED_AT, + description="", + backend_data=f'{{"pod_name": "{JOB_POD_NAME}"}}', + ) + + @pytest.mark.parametrize("primary_id", [JOB_POD_NAME, None]) + def test_deletes_pod_service_and_secret(self, api: Mock, primary_id: Optional[str]): + delete_resource_group({"ctx": _cluster()}, self.group(primary_id)) + + api.delete_namespaced_service.assert_called_once_with( + namespace=NAMESPACE, name=f"{JOB_POD_NAME}-service" + ) + api.delete_namespaced_pod.assert_called_once_with(namespace=NAMESPACE, name=JOB_POD_NAME) + api.delete_namespaced_secret.assert_called_once_with( + namespace=NAMESPACE, name=f"{JOB_POD_NAME}-registry-auth" + ) + + def test_ignores_missing_objects(self, api: Mock): + api.delete_namespaced_service.side_effect = ApiException(status=404) + api.delete_namespaced_pod.side_effect = ApiException(status=404) + api.delete_namespaced_secret.side_effect = ApiException(status=404) + + delete_resource_group({"ctx": _cluster()}, self.group()) + + def test_raises_on_deletion_error(self, api: Mock): + api.delete_namespaced_pod.side_effect = ApiException(status=500) + + with pytest.raises(ComputeError): + delete_resource_group({"ctx": _cluster()}, self.group()) + + api.delete_namespaced_secret.assert_called_once() diff --git a/src/tests/_internal/core/backends/kubernetes/test_utils.py b/src/tests/_internal/core/backends/kubernetes/test_utils.py index f58868217f..ed8c314025 100644 --- a/src/tests/_internal/core/backends/kubernetes/test_utils.py +++ b/src/tests/_internal/core/backends/kubernetes/test_utils.py @@ -1,8 +1,10 @@ import logging from textwrap import dedent from typing import Optional, Union +from unittest.mock import Mock import pytest +from kubernetes import client from dstack._internal.core.backends.kubernetes.models import ( KubeconfigConfig, @@ -13,7 +15,9 @@ from dstack._internal.core.backends.kubernetes.utils import ( Cluster, get_clusters_from_backend_config, + list_all_objects, ) +from dstack._internal.core.errors import ComputeError class TestGetClustersFromBackendConfig: @@ -316,3 +320,32 @@ def test_request_timeout_and_retries_propagate_to_client(self): api_client = clusters[0].api_client assert api_client.configuration.retries == 5 # pyright: ignore[reportAttributeAccessIssue] assert getattr(api_client, "_ApiClient__request_timeout", None) == 7 + + +class TestListAllObjects: + def test_follows_continue_tokens(self): + method = Mock( + side_effect=[ + Mock(items=["a", "b"], metadata=client.V1ListMeta(_continue="next")), + Mock(items=["c"], metadata=client.V1ListMeta(_continue=None)), + ] + ) + + objects = list_all_objects(method, namespace="ns", label_selector="key") + + assert objects == ["a", "b", "c"] + assert [c.kwargs for c in method.call_args_list] == [ + {"namespace": "ns", "label_selector": "key", "limit": 500, "_continue": None}, + {"namespace": "ns", "label_selector": "key", "limit": 500, "_continue": "next"}, + ] + + def test_raises_if_continue_tokens_never_end(self): + method = Mock( + __name__="list_namespaced_pod", + return_value=Mock(items=["a"], metadata=client.V1ListMeta(_continue="next")), + ) + + with pytest.raises(ComputeError, match="list_namespaced_pod.*limit of 1000 pages"): + list_all_objects(method, namespace="ns", label_selector="key") + + assert method.call_count == 1000 From d826eb8e704d8b96366e8055dfa33555d2bd5099 Mon Sep 17 00:00:00 2001 From: Victor Skvortsov Date: Tue, 3 Nov 2026 14:02:11 +0500 Subject: [PATCH 2/2] Document orphan cleanup without permissions to list secrets --- contributing/ORPHAN-CLEANUP.md | 1 + mkdocs/docs/concepts/backends.md | 2 ++ src/dstack/_internal/core/backends/base/compute.py | 2 ++ 3 files changed, 5 insertions(+) diff --git a/contributing/ORPHAN-CLEANUP.md b/contributing/ORPHAN-CLEANUP.md index 1014271b6e..adf7788eab 100644 --- a/contributing/ORPHAN-CLEANUP.md +++ b/contributing/ORPHAN-CLEANUP.md @@ -53,6 +53,7 @@ When instance termination gives up after retrying, the log says that orphan clea - Resources are not cleaned up if the backend credentials change or the backend is deleted. - Compute groups, placement groups, volumes and gateways are not covered. `OwnerType` only has `INSTANCE`. +- A backend may skip internal resources it lacks permissions to list (e.g. Kubernetes secrets without the `list` verb). The cleanup still counts as clean, so such resources left without their owners are found only by cleanups after the permission is granted and the project has new activity. - Resources are recognized as `dstack`'s by the `provider_ref` tag or name alone. A resource that copies them from a `dstack` instance, e.g. a VM created with "Launch more like this" on AWS or "Create similar" on GCP, is classified as an orphan of that instance and deleted in `delete` mode. - `ProviderRefModel` records are never deleted yet. There is one per fleet instance and per placeholder instance, i.e. per job submission that provisions new capacity. - On dstack Sky, each project's `dstack` backend lists the shared base backend accounts, so listings repeat per project. Cleaning up each shared account once is a possible optimization. diff --git a/mkdocs/docs/concepts/backends.md b/mkdocs/docs/concepts/backends.md index 20894f9def..8c783a4b50 100644 --- a/mkdocs/docs/concepts/backends.md +++ b/mkdocs/docs/concepts/backends.md @@ -1371,6 +1371,8 @@ projects: Ensure you've created a ClusterRoleBinding and RoleBinding to grant the roles to the user or the service account you're using. + The `list` verb on `secrets` is only needed for [orphan cleanup](../reference/env.md#DSTACK_SERVER_ORPHAN_CLEANUP_MODE) to find registry auth secrets left without their pods, and `dstack` works without it. Kubernetes can't limit it to `dstack`'s secrets, so it lets `dstack` read the contents of all secrets in the namespace. Grant it only if the namespace is dedicated to `dstack`. + ??? info "Resources and offers" If you use ranges with [`resources`](../concepts/tasks.md#resources) (e.g. `gpu: 1..8` or `memory: 64GB..`) in fleet or run configurations, other backends collect and try all offers that satisfy the range. diff --git a/src/dstack/_internal/core/backends/base/compute.py b/src/dstack/_internal/core/backends/base/compute.py index 2efc68ba50..7616f4bcb7 100644 --- a/src/dstack/_internal/core/backends/base/compute.py +++ b/src/dstack/_internal/core/backends/base/compute.py @@ -494,6 +494,8 @@ def list_resource_groups(self) -> ResourceListing: fails, return no groups of the region and report it in `ResourceListing.failures`. Otherwise, `primary_id=None` may be reported for resources of a tracked owner whose owner resource was not listed, and they would be deleted. + Internal resources that the backend lacks permissions to list may be skipped with + a warning instead, if deleting their group deletes them anyway. Raise if nothing could be listed. """ pass