Support orphan cleanup in the Kubernetes backend

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.

mkdocs/docs/concepts/backends.md

@@ -1394,6 +1394,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.
 

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`.

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"]

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

src/dstack/_internal/core/backends/kubernetes/compute.py

@@ -26,13 +26,15 @@
     ComputeWithInstanceVolumesSupport,
     ComputeWithMultinodeSupport,
     ComputeWithPrivilegedSupport,
+    ComputeWithResourceListingSupport,
     ComputeWithVolumeSupport,
     generate_unique_gateway_instance_name,
     generate_unique_instance_name_for_job,
     generate_unique_name,
     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,6 +43,7 @@
     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 (
     JUMP_POD_DEFAULT_CPU,
     JUMP_POD_DEFAULT_MEMORY,
@@ -50,6 +53,7 @@
     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,
@@ -63,6 +67,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,
@@ -71,6 +76,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,
@@ -86,6 +93,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,
@@ -169,6 +180,7 @@ class KubernetesCompute(
     ComputeWithVolumeSupport,
     ComputeWithGatewaySupport,
     ComputeWithMultinodeSupport,
+    ComputeWithResourceListingSupport,
     Compute,
 ):
     unallocated_resources_argument_has_effect = True
@@ -243,7 +255,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),
@@ -267,6 +279,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,
@@ -277,7 +290,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,
@@ -337,7 +350,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(
@@ -442,7 +455,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)
@@ -485,30 +498,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,
@@ -589,7 +589,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(
@@ -622,7 +622,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)
@@ -1614,25 +1614,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}"