From 55924753da992430db2ac9dff377237bd1d6eeb5 Mon Sep 17 00:00:00 2001 From: Victor Skvortsov Date: Fri, 23 Oct 2026 11:57:54 +0500 Subject: [PATCH 1/6] Support orphaned resources cleanup for RunPod --- .../core/backends/runpod/api_client.py | 33 +++++++++++++++ .../_internal/core/backends/runpod/compute.py | 40 ++++++++++++++++++- src/dstack/_internal/utils/common.py | 11 +++++ 3 files changed, 82 insertions(+), 2 deletions(-) diff --git a/src/dstack/_internal/core/backends/runpod/api_client.py b/src/dstack/_internal/core/backends/runpod/api_client.py index c550cb72da..00b923f1d6 100644 --- a/src/dstack/_internal/core/backends/runpod/api_client.py +++ b/src/dstack/_internal/core/backends/runpod/api_client.py @@ -38,10 +38,13 @@ 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: str class V2PodList(TypedDict): @@ -395,6 +398,9 @@ def get_pod(self, pod_id: str) -> Optional[V2Pod]: raise return response.json() + def list_pods(self) -> list[V2Pod]: + return self._paginate_v2("GET", "pods", response_key="pods") + def terminate_pod(self, pod_id: str) -> bool: try: self._make_v2_request("DELETE", "pods", pod_id) @@ -534,6 +540,33 @@ def _make_graphql_request(self, data: Optional[Dict[str, Any]] = None) -> reques raise BackendInvalidCredentialsError(e.response.text) raise + def _paginate_v2( + self, + method: str, + /, + *path_parts: str, + response_key: str, + json: Optional[Mapping[str, Any]] = None, + ) -> list[Any]: + per_page_limit = 1000 + pagination_pages_hardlimit = 20 + result = [] + for p in range(pagination_pages_hardlimit): + resp = self._make_v2_request( + method, *path_parts, f"?limit={per_page_limit}", json=json + ).json() + batch = resp[response_key] + result.extend(batch) + next_cursor = resp["pagination"]["nextCursor"] + has_next_page = resp["pagination"]["hasNextPage"] + if next_cursor is None or not has_next_page: + break + if p == pagination_pages_hardlimit: + logger.error( + "Pagination exceeded hard limit of %s pages", pagination_pages_hardlimit + ) + return result + def _make_v2_request( self, method: str, /, *path_parts: str, json: Optional[Mapping[str, Any]] = None ) -> requests.Response: diff --git a/src/dstack/_internal/core/backends/runpod/compute.py b/src/dstack/_internal/core/backends/runpod/compute.py index 44dc37032e..8d47b18aa8 100644 --- a/src/dstack/_internal/core/backends/runpod/compute.py +++ b/src/dstack/_internal/core/backends/runpod/compute.py @@ -16,6 +16,7 @@ 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 +25,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 +45,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__) @@ -136,7 +142,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 +410,34 @@ 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: + 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_datetime(pod["createdAt"]), + 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.terminate_instance(group.primary_id, group.region or "") + def register_volume(self, volume: Volume) -> VolumeProvisioningData: assert isinstance(volume.configuration, RunpodVolumeConfiguration) volume_id = get_or_error(volume.configuration.volume_id) diff --git a/src/dstack/_internal/utils/common.py b/src/dstack/_internal/utils/common.py index 33db805f04..8c6f9bd4cd 100644 --- a/src/dstack/_internal/utils/common.py +++ b/src/dstack/_internal/utils/common.py @@ -68,6 +68,17 @@ def get_milliseconds_since_epoch() -> int: return int(round(time.time() * 1000)) +def parse_datetime_or_error(d: str) -> datetime: + return datetime.fromisoformat(d.replace("Z", "+00:00")) + + +def parse_datetime(d: str) -> Optional[datetime]: + try: + return parse_datetime_or_error(d) + except ValueError: + return None + + _DATETIME_ADAPTER = TypeAdapter(datetime) From aaa99fd7d3dc4e200b9df6e2f24be0719385b40d Mon Sep 17 00:00:00 2001 From: Victor Skvortsov Date: Fri, 23 Oct 2026 12:11:13 +0500 Subject: [PATCH 2/6] Fix RunPod orphan cleanup enrollment and pods pagination --- .../core/backends/runpod/api_client.py | 66 ++++++---- .../_internal/core/backends/runpod/compute.py | 16 ++- src/dstack/_internal/utils/common.py | 21 ++- .../core/backends/runpod/test_api_client.py | 34 +++++ .../core/backends/runpod/test_compute.py | 122 +++++++++++++++++- src/tests/_internal/utils/test_common.py | 31 +++++ 6 files changed, 249 insertions(+), 41 deletions(-) diff --git a/src/dstack/_internal/core/backends/runpod/api_client.py b/src/dstack/_internal/core/backends/runpod/api_client.py index 00b923f1d6..9573e5ef59 100644 --- a/src/dstack/_internal/core/backends/runpod/api_client.py +++ b/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 @@ -44,13 +49,18 @@ class V2Pod(TypedDict): mounts: "V2Mounts" cluster: NotRequired["V2PodCluster"] createdAt: str - dataCenterId: 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] @@ -399,7 +409,10 @@ def get_pod(self, pod_id: str) -> Optional[V2Pod]: return response.json() def list_pods(self) -> list[V2Pod]: - return self._paginate_v2("GET", "pods", response_key="pods") + """ + 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: @@ -540,35 +553,35 @@ def _make_graphql_request(self, data: Optional[Dict[str, Any]] = None) -> reques raise BackendInvalidCredentialsError(e.response.text) raise - def _paginate_v2( - self, - method: str, - /, - *path_parts: str, - response_key: str, - json: Optional[Mapping[str, Any]] = None, - ) -> list[Any]: - per_page_limit = 1000 - pagination_pages_hardlimit = 20 + 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 = [] - for p in range(pagination_pages_hardlimit): - resp = self._make_v2_request( - method, *path_parts, f"?limit={per_page_limit}", json=json - ).json() - batch = resp[response_key] - result.extend(batch) - next_cursor = resp["pagination"]["nextCursor"] - has_next_page = resp["pagination"]["hasNextPage"] - if next_cursor is None or not has_next_page: + 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 - if p == pagination_pages_hardlimit: - logger.error( - "Pagination exceeded hard limit of %s pages", pagination_pages_hardlimit - ) + 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( @@ -576,6 +589,7 @@ def _make_v2_request( 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) diff --git a/src/dstack/_internal/core/backends/runpod/compute.py b/src/dstack/_internal/core/backends/runpod/compute.py index 8d47b18aa8..5d5aee373e 100644 --- a/src/dstack/_internal/core/backends/runpod/compute.py +++ b/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,6 +10,7 @@ ComputeWithAllOffersCached, ComputeWithGroupProvisioningSupport, ComputeWithMultinodeSupport, + ComputeWithResourceListingSupport, ComputeWithVolumeSupport, generate_unique_instance_name, generate_unique_volume_name, @@ -75,6 +76,7 @@ class RunpodCompute( ComputeWithVolumeSupport, ComputeWithMultinodeSupport, ComputeWithGroupProvisioningSupport, + ComputeWithResourceListingSupport, Compute, ): _last_cleanup_time = None @@ -414,6 +416,8 @@ 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 @@ -423,7 +427,7 @@ def list_resource_groups(self) -> ResourceListing: provider_ref=provider_ref, primary_id=pod["id"], region=pod["dataCenterId"], - created_at=parse_datetime(pod["createdAt"]), + created_at=_parse_pod_created_at(pod), description=f"pod {pod['id']}", backend_data=None, ) @@ -560,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 diff --git a/src/dstack/_internal/utils/common.py b/src/dstack/_internal/utils/common.py index 8c6f9bd4cd..05717aab54 100644 --- a/src/dstack/_internal/utils/common.py +++ b/src/dstack/_internal/utils/common.py @@ -68,17 +68,6 @@ def get_milliseconds_since_epoch() -> int: return int(round(time.time() * 1000)) -def parse_datetime_or_error(d: str) -> datetime: - return datetime.fromisoformat(d.replace("Z", "+00:00")) - - -def parse_datetime(d: str) -> Optional[datetime]: - try: - return parse_datetime_or_error(d) - except ValueError: - return None - - _DATETIME_ADAPTER = TypeAdapter(datetime) @@ -96,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] diff --git a/src/tests/_internal/core/backends/runpod/test_api_client.py b/src/tests/_internal/core/backends/runpod/test_api_client.py index 712aa3c124..315a57cbdd 100644 --- a/src/tests/_internal/core/backends/runpod/test_api_client.py +++ b/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 diff --git a/src/tests/_internal/core/backends/runpod/test_compute.py b/src/tests/_internal/core/backends/runpod/test_compute.py index dd7e982ddb..964b6128df 100644 --- a/src/tests/_internal/core/backends/runpod/test_compute.py +++ b/src/tests/_internal/core/backends/runpod/test_compute.py @@ -1,13 +1,16 @@ +from datetime import datetime, timezone from typing import Optional from unittest.mock import MagicMock, patch import pytest +from dstack._internal.core.backends.base.compute import parse_provider_ref_from_name from dstack._internal.core.backends.runpod.api_client import RunpodV2ClientError from dstack._internal.core.backends.runpod.compute import RunpodCompute from dstack._internal.core.backends.runpod.models import RunpodAPIKeyCreds, RunpodConfig 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 from dstack._internal.core.models.instances import InstanceOfferWithAvailability from dstack._internal.core.models.resources import ResourcesSpec from dstack._internal.core.models.runs import JobProvisioningData, Requirements @@ -63,14 +66,18 @@ def _run() -> MagicMock: return run -def _run_job(compute: RunpodCompute, offer: InstanceOfferWithAvailability) -> JobProvisioningData: +def _run_job( + compute: RunpodCompute, + offer: InstanceOfferWithAvailability, + job: Optional[MagicMock] = None, +) -> JobProvisioningData: with patch( "dstack._internal.core.backends.runpod.compute.get_docker_commands", return_value=["echo hi"], ): return compute.run_job( run=_run(), - job=_job(), + job=job if job is not None else _job(), instance_offer=offer, project_ssh_public_key=PROJECT_SSH_PUBLIC_KEY, project_ssh_private_key="private-key", @@ -158,6 +165,17 @@ def test_failed_registry_credentials_cleanup_does_not_fail_provisioning(self): assert jpd.instance_id == "pod-1" compute.api_client.terminate_pod.assert_not_called() + def test_encodes_provider_ref_in_pod_name(self): + compute = _compute() + compute.api_client.create_pod.return_value = {"id": "pod-1"} + job = _job() + job.job_submissions = [MagicMock(provider_ref="abcdef123456")] + + _run_job(compute, _gpu_offer(), job=job) + + pod_name = compute.api_client.create_pod.call_args.kwargs["name"] + assert parse_provider_ref_from_name(pod_name) == "abcdef123456" + class TestRunpodComputeRunJobs: def test_sets_internal_ip_from_cluster_ip(self): @@ -256,3 +274,103 @@ def test_cluster_pod_sets_internal_ip(self): assert jpd.hostname == "203.0.113.10" assert jpd.ssh_port == 40022 assert jpd.internal_ip == "10.65.0.3" + + +def _listed_pod( + pod_id: str, + *, + name: str, + status: str = "RUNNING", + data_center_id: Optional[str] = "EU-RO-1", + created_at: str = "2026-10-23T10:00:00Z", +) -> dict: + pod = _pod(pod_id) + pod["status"] = status + pod["name"] = name + pod["dataCenterId"] = data_center_id + pod["createdAt"] = created_at + return pod + + +class TestRunpodComputeListResourceGroups: + def test_lists_pods_with_provider_ref(self): + compute = _compute() + compute.api_client.list_pods.return_value = [ + _listed_pod("pod-1", name="dstack-main-test-job-0-abcdef123456wxyz"), + ] + + listing = compute.list_resource_groups() + + assert listing.failures == [] + assert listing.resource_groups == [ + ListedResourceGroup( + backend=BackendType.RUNPOD, + owner_type=OwnerType.INSTANCE, + provider_ref="abcdef123456", + primary_id="pod-1", + region="EU-RO-1", + created_at=datetime(2026, 10, 23, 10, 0, tzinfo=timezone.utc), + description="pod pod-1", + backend_data=None, + ) + ] + + def test_skips_pods_without_provider_ref(self): + compute = _compute() + compute.api_client.list_pods.return_value = [ + _listed_pod("pod-1", name="dstack-main-test-job-0-abcdefgh"), + _listed_pod("pod-2", name="my-pod"), + ] + + assert compute.list_resource_groups().resource_groups == [] + + def test_skips_terminated_pods(self): + compute = _compute() + compute.api_client.list_pods.return_value = [ + _listed_pod( + "pod-1", name="dstack-main-test-job-0-abcdef123456wxyz", status="TERMINATED" + ), + ] + + assert compute.list_resource_groups().resource_groups == [] + + def test_unparsable_created_at(self): + compute = _compute() + compute.api_client.list_pods.return_value = [ + _listed_pod( + "pod-1", name="dstack-main-test-job-0-abcdef123456wxyz", created_at="invalid" + ), + ] + + [group] = compute.list_resource_groups().resource_groups + + assert group.created_at is None + + +class TestRunpodComputeDeleteResourceGroup: + def _group(self, primary_id: Optional[str]) -> ListedResourceGroup: + return ListedResourceGroup( + backend=BackendType.RUNPOD, + owner_type=OwnerType.INSTANCE, + provider_ref="abcdef123456", + primary_id=primary_id, + region="EU-RO-1", + created_at=None, + description="pod pod-1", + backend_data=None, + ) + + def test_terminates_pod(self): + compute = _compute() + + compute.delete_resource_group(self._group("pod-1")) + + compute.api_client.terminate_pod.assert_called_once_with("pod-1") + + def test_pod_already_deleted(self): + compute = _compute() + compute.api_client.terminate_pod.return_value = False + + compute.delete_resource_group(self._group("pod-1")) + + compute.api_client.terminate_pod.assert_called_once_with("pod-1") diff --git a/src/tests/_internal/utils/test_common.py b/src/tests/_internal/utils/test_common.py index 91f8bf0768..d78c1b143e 100644 --- a/src/tests/_internal/utils/test_common.py +++ b/src/tests/_internal/utils/test_common.py @@ -12,9 +12,11 @@ has_duplicates, local_time, make_proxy_url, + parse_datetime, parse_memory, pretty_date, pretty_resources, + render_datetime_as_api, sizeof_fmt, ) @@ -151,6 +153,35 @@ def test_has_duplicates(iterable, expected): assert has_duplicates(iterable) == expected +class TestParseDatetime: + @pytest.mark.parametrize( + ("value", "expected"), + [ + ("2026-10-23T10:00:00Z", datetime(2026, 10, 23, 10, 0, tzinfo=timezone.utc)), + ( + "2026-10-23T10:00:00.12345Z", + datetime(2026, 10, 23, 10, 0, 0, 123450, tzinfo=timezone.utc), + ), + ( + "2026-10-23T12:00:00+02:00", + datetime(2026, 10, 23, 10, 0, tzinfo=timezone.utc), + ), + ("2026-10-23T10:00:00", datetime(2026, 10, 23, 10, 0)), + ], + ) + def test_parses(self, value: str, expected: datetime) -> None: + assert parse_datetime(value) == expected + + def test_roundtrips_render_datetime_as_api(self) -> None: + dt = datetime(2026, 10, 23, 10, 0, 0, 123456, tzinfo=timezone.utc) + assert parse_datetime(render_datetime_as_api(dt)) == dt + + @pytest.mark.parametrize("value", ["invalid", "", "2026-13-01T00:00:00Z"]) + def test_raises_value_error(self, value: str) -> None: + with pytest.raises(ValueError): + parse_datetime(value) + + class TestParseMemory: @pytest.mark.parametrize( "memory,as_units,expected", From 31d8927795937ccd8e42233e7bd32fa52815973d Mon Sep 17 00:00:00 2001 From: Victor Skvortsov Date: Fri, 23 Oct 2026 12:14:20 +0500 Subject: [PATCH 3/6] Delete RunPod orphan pods without a region --- src/dstack/_internal/core/backends/runpod/compute.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/dstack/_internal/core/backends/runpod/compute.py b/src/dstack/_internal/core/backends/runpod/compute.py index 5d5aee373e..324a74a9a6 100644 --- a/src/dstack/_internal/core/backends/runpod/compute.py +++ b/src/dstack/_internal/core/backends/runpod/compute.py @@ -440,7 +440,7 @@ def list_resource_groups(self) -> ResourceListing: def delete_resource_group(self, group: ListedResourceGroup) -> None: if group.primary_id is None: return - self.terminate_instance(group.primary_id, group.region or "") + self.api_client.terminate_pod(group.primary_id) def register_volume(self, volume: Volume) -> VolumeProvisioningData: assert isinstance(volume.configuration, RunpodVolumeConfiguration) From 5fc4163dac58e799bd4293b1155c5ca04f1e3705 Mon Sep 17 00:00:00 2001 From: Victor Skvortsov Date: Fri, 23 Oct 2026 12:18:33 +0500 Subject: [PATCH 4/6] Use parse_datetime for API, logs and CLI datetimes --- .../_internal/cli/services/presets/output.py | 7 ++----- .../server/services/logs/fluentbit.py | 5 ++--- src/dstack/_internal/utils/common.py | 2 +- src/tests/_internal/utils/test_common.py | 21 +++++++++++++++++++ 4 files changed, 26 insertions(+), 9 deletions(-) diff --git a/src/dstack/_internal/cli/services/presets/output.py b/src/dstack/_internal/cli/services/presets/output.py index 52df677d61..86f39d54c7 100644 --- a/src/dstack/_internal/cli/services/presets/output.py +++ b/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 = "" diff --git a/src/dstack/_internal/server/services/logs/fluentbit.py b/src/dstack/_internal/server/services/logs/fluentbit.py index bb97e21f09..f49f56e9a2 100644 --- a/src/dstack/_internal/server/services/logs/fluentbit.py +++ b/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: diff --git a/src/dstack/_internal/utils/common.py b/src/dstack/_internal/utils/common.py index 05717aab54..9f0f4208ca 100644 --- a/src/dstack/_internal/utils/common.py +++ b/src/dstack/_internal/utils/common.py @@ -212,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: diff --git a/src/tests/_internal/utils/test_common.py b/src/tests/_internal/utils/test_common.py index d78c1b143e..789b4854e5 100644 --- a/src/tests/_internal/utils/test_common.py +++ b/src/tests/_internal/utils/test_common.py @@ -14,6 +14,7 @@ make_proxy_url, parse_datetime, parse_memory, + parse_since, pretty_date, pretty_resources, render_datetime_as_api, @@ -182,6 +183,26 @@ def test_raises_value_error(self, value: str) -> None: parse_datetime(value) +class TestParseSince: + @pytest.mark.time_machine(datetime(2023, 10, 4, 12, 0, tzinfo=timezone.utc), tick=False) + def test_duration(self) -> None: + assert parse_since("5m") == datetime(2023, 10, 4, 11, 55, tzinfo=timezone.utc) + + @pytest.mark.parametrize( + "value", ["2023-09-24T15:30:00Z", "2023-09-24T15:30:00+00:00", "2023-09-24T17:30:00+02:00"] + ) + def test_datetime(self, value: str) -> None: + assert parse_since(value) == datetime(2023, 9, 24, 15, 30, tzinfo=timezone.utc) + + def test_naive_datetime(self) -> None: + with pytest.raises(ValueError): + parse_since("2023-09-24T15:30:00") + + def test_invalid(self) -> None: + with pytest.raises(ValueError, match="Invalid datetime format"): + parse_since("invalid") + + class TestParseMemory: @pytest.mark.parametrize( "memory,as_units,expected", From e2d3dd343857e1a032ef9aaca482439c0975e134 Mon Sep 17 00:00:00 2001 From: Victor Skvortsov Date: Fri, 23 Oct 2026 12:41:54 +0500 Subject: [PATCH 5/6] Make orphan cleanup enrollment interval configurable --- .../_internal/server/background/scheduled_tasks/__init__.py | 4 +++- src/dstack/_internal/server/settings.py | 6 ++++-- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/src/dstack/_internal/server/background/scheduled_tasks/__init__.py b/src/dstack/_internal/server/background/scheduled_tasks/__init__.py index 46408d2602..33aa278303 100644 --- a/src/dstack/_internal/server/background/scheduled_tasks/__init__.py +++ b/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( diff --git a/src/dstack/_internal/server/settings.py b/src/dstack/_internal/server/settings.py index 806c45b53b..0b04a51186 100644 --- a/src/dstack/_internal/server/settings.py +++ b/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""" From 39ae5bbd7f8d2698eafbda848984460ae1ad02f4 Mon Sep 17 00:00:00 2001 From: Victor Skvortsov Date: Fri, 23 Oct 2026 13:56:33 +0500 Subject: [PATCH 6/6] Document that orphan reports are not deduplicated --- contributing/ORPHAN-CLEANUP.md | 2 +- .../server/background/pipeline_tasks/orphan_cleanup.py | 3 +++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/contributing/ORPHAN-CLEANUP.md b/contributing/ORPHAN-CLEANUP.md index 5c04d67d4f..c720aef344 100644 --- a/contributing/ORPHAN-CLEANUP.md +++ b/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: diff --git a/src/dstack/_internal/server/background/pipeline_tasks/orphan_cleanup.py b/src/dstack/_internal/server/background/pipeline_tasks/orphan_cleanup.py index 1a401bb5f0..8406b1a664 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/orphan_cleanup.py +++ b/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,