Support orphaned resources cleanup for RunPod

contributing/ORPHAN-CLEANUP.md

@@ -35,7 +35,7 @@ Processing a row:
     - No record: ignored. Resources of other servers and projects are never touched.
 - STEP 5: Owners that were provisioned or terminated around the listing are skipped instead of classified as orphans, since the listing may predate the change. The margin covers clock skew between replicas, the lag between setting timestamps and committing them, and providers listing new resources with a delay.
 - STEP 6: Handle orphans according to `DSTACK_SERVER_ORPHAN_CLEANUP_MODE`:
-    - `report` (default): log, send to Sentry, and emit an event targeting the owner instance. If the instance was deleted, the event still targets it by the id and name kept in `ProviderRefModel`, so it shows next to the instance's earlier events.
+    - `report` (default): log, send to Sentry, and emit an event targeting the owner instance. If the instance was deleted, the event still targets it by the id and name kept in `ProviderRefModel`, so it shows next to the instance's earlier events. Reports are not deduplicated: an orphan that stays is reported again on every cleanup of the backend, at most once per cleanup interval. This is expected and accepted.
     - `delete`: call `delete_resource_group()` and emit an event. If the backend raises `NotYetTerminated` for a group with an owner resource, the owner's deletion was requested and the event says the resources are being deleted. The remaining internal resources are listed and deleted by next cleanups.
     - `off`: the pipeline and enrollment are disabled.
 - STEP 7: Set the outcome:

src/dstack/_internal/cli/services/presets/output.py

@@ -1,5 +1,4 @@
 from collections import defaultdict
-from datetime import datetime
 from typing import Any, Optional, Sequence
 
 from rich.table import Table
@@ -8,7 +7,7 @@
 from dstack._internal.cli.utils.common import add_row_from_dict, console
 from dstack._internal.core.models.configurations import PresetConfiguration
 from dstack._internal.core.models.presets import PresetWorkload
-from dstack._internal.utils.common import pretty_date, pretty_resources
+from dstack._internal.utils.common import parse_datetime, pretty_date, pretty_resources
 
 _STATUS_DISPLAY = {
     "verified": ("verified", "secondary"),
@@ -161,9 +160,7 @@ def _add_session(table: Table, session: dict[str, Any], *, verbose: bool = False
     created_at = session.get("created_at")
     if isinstance(created_at, str):
         try:
-            # `Z` is what pydantic v2 emits for UTC, and `datetime.fromisoformat` only
-            # accepts it from Python 3.11; dstack supports 3.10.
-            created = pretty_date(datetime.fromisoformat(created_at.replace("Z", "+00:00")))
+            created = pretty_date(parse_datetime(created_at))
         except ValueError:
             created = created_at
     benchmark = ""

src/dstack/_internal/core/backends/runpod/api_client.py

@@ -19,6 +19,11 @@
 
 AUTH_ERROR_STATUS_CODES = (HTTPStatus.UNAUTHORIZED, HTTPStatus.FORBIDDEN)
 
+PAGINATION_PAGE_LIMIT = 1000
+"""The maximum page size of paginated v2 endpoints."""
+
+PAGINATION_MAX_PAGES = 20
+
 # Applies to non-JSON error responses, which are not expected to contain anything useful past
 # the first few lines
 MAX_ERROR_DETAIL_LEN = 200
@@ -38,16 +43,24 @@ class V2ErrorResponse(TypedDict):
 
 class V2Pod(TypedDict):
     id: str
+    name: str
     status: Literal["PROVISIONING", "STARTING", "RUNNING", "EXITED", "ERROR", "TERMINATED"]
     runtime: Optional["V2PodRuntime"]
     mounts: "V2Mounts"
     cluster: NotRequired["V2PodCluster"]
+    createdAt: str
+    dataCenterId: Optional[str]
 
 
 class V2PodList(TypedDict):
     pods: list[V2Pod]
 
 
+class V2Pagination(TypedDict):
+    nextCursor: Optional[str]
+    hasNextPage: bool
+
+
 class V2PodCluster(TypedDict):
     id: str
     rank: Optional[int]
@@ -395,6 +408,12 @@ def get_pod(self, pod_id: str) -> Optional[V2Pod]:
             raise
         return response.json()
 
+    def list_pods(self) -> list[V2Pod]:
+        """
+        Lists pods except cluster member pods. Deleted pods are not listed.
+        """
+        return self._paginate_v2("pods", response_key="pods")
+
     def terminate_pod(self, pod_id: str) -> bool:
         try:
             self._make_v2_request("DELETE", "pods", pod_id)
@@ -534,15 +553,43 @@ def _make_graphql_request(self, data: Optional[Dict[str, Any]] = None) -> reques
                 raise BackendInvalidCredentialsError(e.response.text)
             raise
 
+    def _paginate_v2(self, *path_parts: str, response_key: str) -> list[Any]:
+        """
+        Follows `pagination.nextCursor` of a cursor-paginated GET endpoint and returns
+        the items of all pages, or of the first `PAGINATION_MAX_PAGES` pages.
+        """
+        result = []
+        params: dict[str, Any] = {"limit": PAGINATION_PAGE_LIMIT}
+        for _ in range(PAGINATION_MAX_PAGES):
+            response_json = self._make_v2_request("GET", *path_parts, params=params).json()
+            result.extend(response_json[response_key])
+            pagination: V2Pagination = response_json["pagination"]
+            if not pagination["hasNextPage"] or pagination["nextCursor"] is None:
+                break
+            params["cursor"] = pagination["nextCursor"]
+        else:
+            logger.error(
+                "Listing %s exceeded the limit of %s pages, the rest is skipped",
+                "/".join(path_parts),
+                PAGINATION_MAX_PAGES,
+            )
+        return result
+
     def _make_v2_request(
-        self, method: str, /, *path_parts: str, json: Optional[Mapping[str, Any]] = None
+        self,
+        method: str,
+        /,
+        *path_parts: str,
+        json: Optional[Mapping[str, Any]] = None,
+        params: Optional[Mapping[str, Any]] = None,
     ) -> requests.Response:
         url = "/".join((API_V2_URL, *path_parts))
         response = self._session.request(
             method=method,
             url=url,
             timeout=REQUEST_TIMEOUT,
             json=json,
+            params=params,
         )
         if HTTPStatus.BAD_REQUEST <= response.status_code < HTTPStatus.INTERNAL_SERVER_ERROR:
             error = _parse_v2_error_response(response)

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

@@ -1,7 +1,7 @@
 import json
 import re
 import uuid
-from datetime import timedelta
+from datetime import datetime, timedelta
 from typing import Callable, List, Optional
 
 from dstack._internal.core.backends.base.authorized_keys import build_authorized_keys
@@ -10,12 +10,14 @@
     ComputeWithAllOffersCached,
     ComputeWithGroupProvisioningSupport,
     ComputeWithMultinodeSupport,
+    ComputeWithResourceListingSupport,
     ComputeWithVolumeSupport,
     generate_unique_instance_name,
     generate_unique_volume_name,
     get_docker_commands,
     get_job_instance_name,
     get_job_provider_ref,
+    parse_provider_ref_from_name,
 )
 from dstack._internal.core.backends.base.models import JobConfiguration
 from dstack._internal.core.backends.base.offers import get_catalog_offers
@@ -24,6 +26,11 @@
 from dstack._internal.core.consts import DSTACK_RUNNER_SSH_PORT
 from dstack._internal.core.errors import BackendError, ComputeError
 from dstack._internal.core.models.backends.base import BackendType
+from dstack._internal.core.models.backends.orphans import (
+    ListedResourceGroup,
+    OwnerType,
+    ResourceListing,
+)
 from dstack._internal.core.models.common import CoreModel, RegistryAuth, validate_extra_ignore
 from dstack._internal.core.models.compute_groups import ComputeGroup, ComputeGroupProvisioningData
 from dstack._internal.core.models.instances import (
@@ -39,7 +46,7 @@
     Volume,
     VolumeProvisioningData,
 )
-from dstack._internal.utils.common import get_current_datetime, get_or_error
+from dstack._internal.utils.common import get_current_datetime, get_or_error, parse_datetime
 from dstack._internal.utils.logging import get_logger
 
 logger = get_logger(__name__)
@@ -69,6 +76,7 @@ class RunpodCompute(
     ComputeWithVolumeSupport,
     ComputeWithMultinodeSupport,
     ComputeWithGroupProvisioningSupport,
+    ComputeWithResourceListingSupport,
     Compute,
 ):
     _last_cleanup_time = None
@@ -136,7 +144,9 @@ def run_job(
             provider_ref=get_job_provider_ref(job),
         )
 
-        pod_name = generate_unique_instance_name(instance_config, max_length=MAX_RESOURCE_NAME_LEN)
+        pod_name = generate_unique_instance_name(
+            instance_config, max_length=MAX_RESOURCE_NAME_LEN, with_provider_ref=True
+        )
         authorized_keys = build_authorized_keys(project_ssh_public_key, extra_authorized_keys)
 
         network_volume_id = None
@@ -402,6 +412,36 @@ def update_provisioning_data(
                 provisioning_data.hostname = port.get("ip")
                 provisioning_data.ssh_port = port.get("public")
 
+    def list_resource_groups(self) -> ResourceListing:
+        resource_groups = []
+        pods = self.api_client.list_pods()
+        for pod in pods:
+            if pod["status"] == "TERMINATED":
+                continue
+            provider_ref = parse_provider_ref_from_name(pod["name"])
+            if provider_ref is None:
+                continue
+            resource_group = ListedResourceGroup(
+                backend=BackendType.RUNPOD,
+                owner_type=OwnerType.INSTANCE,
+                provider_ref=provider_ref,
+                primary_id=pod["id"],
+                region=pod["dataCenterId"],
+                created_at=_parse_pod_created_at(pod),
+                description=f"pod {pod['id']}",
+                backend_data=None,
+            )
+            resource_groups.append(resource_group)
+        return ResourceListing(
+            resource_groups=resource_groups,
+            failures=[],
+        )
+
+    def delete_resource_group(self, group: ListedResourceGroup) -> None:
+        if group.primary_id is None:
+            return
+        self.api_client.terminate_pod(group.primary_id)
+
     def register_volume(self, volume: Volume) -> VolumeProvisioningData:
         assert isinstance(volume.configuration, RunpodVolumeConfiguration)
         volume_id = get_or_error(volume.configuration.volume_id)
@@ -524,3 +564,11 @@ def _get_pod_cluster_ip(pod: V2Pod) -> Optional[str]:
     if cluster is None:
         return None
     return cluster.get("ip")
+
+
+def _parse_pod_created_at(pod: V2Pod) -> Optional[datetime]:
+    try:
+        return parse_datetime(pod["createdAt"])
+    except ValueError:
+        logger.warning("Failed to parse createdAt %r of pod %s", pod["createdAt"], pod["id"])
+        return None

src/dstack/_internal/server/background/pipeline_tasks/orphan_cleanup.py

@@ -552,6 +552,9 @@ async def _handle_orphaned_resource_group(
     resource_group: ListedResourceGroup,
 ) -> _ResourceGroupDecision:
     if settings.ORPHAN_CLEANUP_MODE == OrphanCleanupMode.REPORT:
+        # Expected and accepted: reports are not deduplicated, so an orphan that stays
+        # is reported (logged, sent to Sentry, emitted as an event) on every cleanup
+        # of the backend, i.e. at most once per cleanup interval while the backend is eligible.
         logger.info(
             "Found orphaned %s %s in %s/%s",
             resource_group.owner_type.value,

src/dstack/_internal/server/background/scheduled_tasks/__init__.py

@@ -59,7 +59,9 @@ def start_scheduled_tasks() -> AsyncIOScheduler:
     _scheduler.add_job(delete_instance_healthchecks, IntervalTrigger(minutes=5), max_instances=1)
     if settings.ORPHAN_CLEANUP_MODE != OrphanCleanupMode.OFF:
         _scheduler.add_job(
-            enroll_backends_in_orphan_cleanup, IntervalTrigger(minutes=10), max_instances=1
+            enroll_backends_in_orphan_cleanup,
+            IntervalTrigger(seconds=settings.ORPHAN_CLEANUP_ENROLLMENT_INTERVAL_SECONDS),
+            max_instances=1,
         )
     if settings.ENABLE_PROMETHEUS_METRICS:
         _scheduler.add_job(

src/dstack/_internal/server/services/logs/fluentbit.py

@@ -1,4 +1,3 @@
-from datetime import datetime
 from typing import List, Optional, Protocol
 from uuid import UUID
 
@@ -19,7 +18,7 @@
     LogStorageError,
     unix_time_ms_to_datetime,
 )
-from dstack._internal.utils.common import batched
+from dstack._internal.utils.common import batched, parse_datetime
 from dstack._internal.utils.logging import get_logger
 
 logger = get_logger(__name__)
@@ -113,7 +112,7 @@ def read(
 
                 if timestamp_str:
                     try:
-                        timestamp = datetime.fromisoformat(timestamp_str.replace("Z", "+00:00"))
+                        timestamp = parse_datetime(timestamp_str)
                     except ValueError:
                         continue
                 else:

src/dstack/_internal/server/settings.py

@@ -240,9 +240,11 @@ class JobNetworkMode(Enum):
 ORPHAN_CLEANUP_INTERVAL_SECONDS = environ.get_int(
     "DSTACK_SERVER_ORPHAN_CLEANUP_INTERVAL_SECONDS", default=3600
 )
-
-
 DEFAULT_ORPHAN_CLEANUP_MODE = OrphanCleanupMode.REPORT
 ORPHAN_CLEANUP_MODE = environ.get_enum(
     "DSTACK_SERVER_ORPHAN_CLEANUP_MODE", OrphanCleanupMode, default=DEFAULT_ORPHAN_CLEANUP_MODE
 )
+ORPHAN_CLEANUP_ENROLLMENT_INTERVAL_SECONDS = environ.get_int(
+    "DSTACK_SERVER_ORPHAN_CLEANUP_ENROLLMENT_INTERVAL_SECONDS", default=600
+)
+"""Internal, e.g. for e2e tests"""

src/dstack/_internal/utils/common.py

@@ -85,6 +85,16 @@ def render_datetime_as_api(value: datetime) -> str:
     return _DATETIME_ADAPTER.dump_python(value, mode="json")
 
 
+def parse_datetime(value: str) -> datetime:
+    """
+    Parse a datetime the way the API parses one: an ISO 8601 string or a Unix timestamp.
+
+    Unlike `datetime.fromisoformat()` on Python 3.10, accepts the `Z` suffix and fractional
+    seconds of any precision. Raises `ValueError` if `value` is not a datetime.
+    """
+    return _DATETIME_ADAPTER.validate_strings(value)
+
+
 DateFormatter = Callable[[datetime], str]
 
 
@@ -202,7 +212,7 @@ def parse_since(value: str) -> datetime:
     except ValueError:
         pass
     try:
-        res = datetime.fromisoformat(value)
+        res = parse_datetime(value)
     except ValueError:
         raise ValueError("Invalid datetime format")
     else:

src/tests/_internal/core/backends/runpod/test_api_client.py

@@ -2,6 +2,7 @@
 
 from dstack._internal.core.backends.runpod.api_client import (
     API_V2_URL,
+    PAGINATION_MAX_PAGES,
     RunpodApiClient,
     RunpodV2ClientError,
 )
@@ -193,3 +194,36 @@ def test_other_errors_are_raised(self, requests_mock):
 
         with pytest.raises(RunpodV2ClientError):
             _client().terminate_pod("pod-1")
+
+
+class TestRunpodApiClientListPods:
+    def _page(self, pod_ids: list[str], next_cursor: str | None) -> dict:
+        return {
+            "pods": [{"id": pod_id} for pod_id in pod_ids],
+            "pagination": {"nextCursor": next_cursor, "hasNextPage": next_cursor is not None},
+        }
+
+    def test_follows_cursor(self, requests_mock):
+        requests_mock.get(
+            f"{API_V2_URL}/pods",
+            [
+                {"json": self._page(["pod-1", "pod-2"], "cursor-1")},
+                {"json": self._page(["pod-3"], None)},
+            ],
+        )
+
+        pods = _client().list_pods()
+
+        assert [pod["id"] for pod in pods] == ["pod-1", "pod-2", "pod-3"]
+        assert [r.qs for r in requests_mock.request_history] == [
+            {"limit": ["1000"]},
+            {"limit": ["1000"], "cursor": ["cursor-1"]},
+        ]
+
+    def test_stops_at_max_pages(self, requests_mock):
+        requests_mock.get(f"{API_V2_URL}/pods", json=self._page(["pod-1"], "cursor-1"))
+
+        pods = _client().list_pods()
+
+        assert len(pods) == PAGINATION_MAX_PAGES
+        assert requests_mock.call_count == PAGINATION_MAX_PAGES