diff --git a/contributing/ORPHAN-CLEANUP.md b/contributing/ORPHAN-CLEANUP.md index c720aef34..1014271b6 100644 --- a/contributing/ORPHAN-CLEANUP.md +++ b/contributing/ORPHAN-CLEANUP.md @@ -53,5 +53,6 @@ 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`. +- 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 e488520fb..a7bf1019b 100644 --- a/mkdocs/docs/concepts/backends.md +++ b/mkdocs/docs/concepts/backends.md @@ -545,6 +545,7 @@ gcloud projects list --format="json(projectId)" compute.instances.delete compute.instances.detachDisk compute.instances.get + compute.instances.list compute.instances.setLabels compute.instances.setMetadata compute.instances.setServiceAccount @@ -605,6 +606,7 @@ gcloud projects list --format="json(projectId)" ``` tpu.nodes.create tpu.nodes.get + tpu.nodes.list tpu.nodes.update tpu.nodes.delete tpu.operations.get diff --git a/mkdocs/docs/reference/env.md b/mkdocs/docs/reference/env.md index 09082e084..88279f089 100644 --- a/mkdocs/docs/reference/env.md +++ b/mkdocs/docs/reference/env.md @@ -151,7 +151,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. Defaults to `report`. Currently supported for the `aws` backend. +- `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`, `gcp`, 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/src/dstack/_internal/core/backends/gcp/compute.py b/src/dstack/_internal/core/backends/gcp/compute.py index 9cd795e68..273e8bd63 100644 --- a/src/dstack/_internal/core/backends/gcp/compute.py +++ b/src/dstack/_internal/core/backends/gcp/compute.py @@ -2,8 +2,9 @@ import json import re from collections import defaultdict -from collections.abc import Iterable +from collections.abc import Iterable, Mapping from dataclasses import dataclass +from datetime import datetime from typing import Callable, Dict, List, Literal, Optional, Tuple import google.api_core.exceptions @@ -19,6 +20,7 @@ import dstack._internal.core.backends.gcp.resources as gcp_resources from dstack._internal import settings from dstack._internal.core.backends.base.compute import ( + PROVIDER_REF_TAG_KEY, Compute, ComputeTTLCache, ComputeWithAllOffersCached, @@ -31,6 +33,7 @@ ComputeWithPrivateGatewaySupport, ComputeWithPrivilegedSupport, ComputeWithReservationSupport, + ComputeWithResourceListingSupport, ComputeWithVolumeSupport, generate_unique_gateway_instance_name, generate_unique_instance_name, @@ -39,6 +42,7 @@ get_gateway_user_data, get_shim_commands, get_user_data, + is_valid_provider_ref, merge_tags, requires_nvidia_proprietary_kernel_modules, ) @@ -57,6 +61,12 @@ ProvisioningError, ) 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_extra_ignore, @@ -84,7 +94,7 @@ VolumeAttachmentData, VolumeProvisioningData, ) -from dstack._internal.utils.common import get_or_error +from dstack._internal.utils.common import get_or_error, parse_datetime from dstack._internal.utils.logging import get_logger logger = get_logger(__name__) @@ -121,6 +131,11 @@ class GCPGatewayBackendData(CoreModel): forwarding_rule_name: str +class GCPResourceGroupBackendData(CoreModel): + zone: str + is_tpu: bool = False + + class GCPCompute( ComputeWithAllOffersCached, ComputeWithCreateInstanceSupport, @@ -133,6 +148,7 @@ class GCPCompute( ComputeWithPrivateGatewaySupport, ComputeWithGatewayLoadBalancerSupport, ComputeWithVolumeSupport, + ComputeWithResourceListingSupport, Compute, ): def __init__(self, config: GCPConfig): @@ -278,19 +294,51 @@ def terminate_instance( backend_data_dict = json.loads(backend_data) zone = backend_data_dict["zone"] is_tpu = backend_data_dict.get("is_tpu", False) - try: - if is_tpu: - name = f"projects/{self.config.project_id}/locations/{zone}/nodes/{instance_id}" - delete_request = tpu_v2.DeleteNodeRequest(name=name) - self.tpu_client.delete_node(request=delete_request) - else: - self.instances_client.delete( - project=self.config.project_id, - zone=zone, - instance=instance_id, + self._delete_instance(instance_id=instance_id, zone=zone, is_tpu=is_tpu) + + def list_resource_groups(self) -> ResourceListing: + regions = get_or_error(self.config.regions) + resource_groups, unreachable_zones = self._list_vm_resource_groups() + # TPUs are not listed if disabled so that the TPU API and permissions stay optional. + if self.config.allow_tpu: + tpu_resource_groups, tpu_unreachable_zones = self._list_tpu_resource_groups() + resource_groups.extend(tpu_resource_groups) + unreachable_zones.extend(tpu_unreachable_zones) + unreachable_zones_by_region: Dict[str, List[str]] = defaultdict(list) + for zone in unreachable_zones: + unreachable_zones_by_region[_get_zone_region(zone)].append(zone) + listing = ResourceListing( + resource_groups=[ + g + for g in resource_groups + if g.region in regions and g.region not in unreachable_zones_by_region + ], + failures=[ + ResourceListingFailure( + backend=BackendType.GCP, + region=region, + message=f"Zones unreachable: {', '.join(sorted(set(zones)))}", ) - except google.api_core.exceptions.NotFound: - pass + for region, zones in unreachable_zones_by_region.items() + if region in regions + ], + ) + if len(listing.failures) == len(regions): + raise ComputeError(f"Failed to list resources in all regions: {listing.failures}") + return listing + + def delete_resource_group(self, group: ListedResourceGroup) -> None: + backend_data = validate_json_extra_ignore( + GCPResourceGroupBackendData, get_or_error(group.backend_data) + ) + try: + self._delete_instance( + instance_id=get_or_error(group.primary_id), + zone=backend_data.zone, + is_tpu=backend_data.is_tpu, + ) + except google.api_core.exceptions.GoogleAPICallError as e: + raise ComputeError(f"Failed to delete resource group {group.provider_ref}: {e}") from e def create_instance( self, @@ -340,6 +388,8 @@ def create_instance( "dstack_name": instance_config.instance_name, "dstack_user": instance_config.user.lower(), } + if instance_config.provider_ref is not None: + labels[PROVIDER_REF_TAG_KEY] = instance_config.provider_ref labels = merge_tags( base_tags=labels, backend_tags=self.config.tags, @@ -1453,6 +1503,155 @@ def _find_reservation(self, configured_name: str) -> dict[str, compute_v1.Reserv name=name, ) + def _delete_instance(self, instance_id: str, zone: str, is_tpu: bool) -> None: + try: + if is_tpu: + name = f"projects/{self.config.project_id}/locations/{zone}/nodes/{instance_id}" + delete_request = tpu_v2.DeleteNodeRequest(name=name) + self.tpu_client.delete_node(request=delete_request) + else: + self.instances_client.delete( + project=self.config.project_id, + zone=zone, + instance=instance_id, + ) + except google.api_core.exceptions.NotFound: + pass + + def _list_vm_resource_groups(self) -> Tuple[List[ListedResourceGroup], List[str]]: + """ + Lists VMs labeled with `provider_ref` in all zones. + Returns the resource groups and the unreachable zones. + """ + resource_groups = [] + unreachable_zones = [] + request = compute_v1.AggregatedListInstancesRequest( + project=self.config.project_id, + filter=f"labels.{PROVIDER_REF_TAG_KEY}:*", + ) + try: + for page in self.instances_client.aggregated_list(request=request).pages: + unreachable_zones.extend(_get_last_path_segment(s) for s in page.unreachables) + for scope, scoped_list in page.items.items(): + zone = _get_last_path_segment(scope) + if scoped_list.warning.code == "UNREACHABLE": + unreachable_zones.append(zone) + for instance in scoped_list.instances: + resource_group = _get_vm_resource_group(instance, zone) + if resource_group is not None: + resource_groups.append(resource_group) + except google.api_core.exceptions.Forbidden as e: + raise ComputeError( + f"Failed to list instances. Grant the compute.instances.list permission: {e}" + ) from e + except google.api_core.exceptions.GoogleAPICallError as e: + raise ComputeError(f"Failed to list instances: {e}") from e + return resource_groups, unreachable_zones + + def _list_tpu_resource_groups(self) -> Tuple[List[ListedResourceGroup], List[str]]: + """ + Lists TPU nodes labeled with `provider_ref` in all zones. + Returns the resource groups and the unreachable zones. + """ + resource_groups = [] + unreachable_zones = [] + request = tpu_v2.ListNodesRequest(parent=f"projects/{self.config.project_id}/locations/-") + try: + for page in self.tpu_client.list_nodes(request=request).pages: + unreachable_zones.extend(_get_last_path_segment(loc) for loc in page.unreachable) + for node in page.nodes: + resource_group = _get_tpu_resource_group(node) + if resource_group is not None: + resource_groups.append(resource_group) + except google.api_core.exceptions.Forbidden as e: + raise ComputeError( + "Failed to list TPU nodes. Enable the TPU API and grant" + f" the tpu.nodes.list permission: {e}" + ) from e + except google.api_core.exceptions.GoogleAPICallError as e: + raise ComputeError(f"Failed to list TPU nodes: {e}") from e + return resource_groups, unreachable_zones + + +def _get_vm_resource_group(instance: Instance, zone: str) -> Optional[ListedResourceGroup]: + # dstack never stops VMs, so a stopping VM is being deleted or preempted. + # A preempted VM ends up TERMINATED and is listed by next cleanups. + if instance.status == "STOPPING": + return None + provider_ref = _get_resource_provider_ref(instance.labels, instance.name) + if provider_ref is None: + return None + machine_type = _get_last_path_segment(instance.machine_type) + return ListedResourceGroup( + backend=BackendType.GCP, + owner_type=OwnerType.INSTANCE, + provider_ref=provider_ref, + primary_id=instance.name, + region=_get_zone_region(zone), + created_at=_parse_instance_creation_timestamp(instance), + description=f"instance {instance.name} ({machine_type}, {instance.status})", + backend_data=GCPResourceGroupBackendData(zone=zone).model_dump_json(), + ) + + +def _get_tpu_resource_group(node: tpu_v2.Node) -> Optional[ListedResourceGroup]: + if node.state == tpu_v2.Node.State.DELETING: + return None + # projects/{project}/locations/{zone}/nodes/{node_id} + _, _, _, zone, _, node_id = node.name.split("/") + provider_ref = _get_resource_provider_ref(node.labels, node_id) + if provider_ref is None: + return None + return ListedResourceGroup( + backend=BackendType.GCP, + owner_type=OwnerType.INSTANCE, + provider_ref=provider_ref, + primary_id=node_id, + region=_get_zone_region(zone), + created_at=node.create_time, # pyright: ignore[reportArgumentType] + description=f"TPU node {node_id} ({node.accelerator_type}, {node.state.name})", + backend_data=GCPResourceGroupBackendData(zone=zone, is_tpu=True).model_dump_json(), + ) + + +def _get_resource_provider_ref(labels: Mapping[str, str], resource_name: str) -> Optional[str]: + provider_ref = labels.get(PROVIDER_REF_TAG_KEY) + if provider_ref is None: + return None + if not is_valid_provider_ref(provider_ref): + logger.warning( + "Skipping GCP resource %s with invalid %s label value %r", + resource_name, + PROVIDER_REF_TAG_KEY, + provider_ref, + ) + return None + return provider_ref + + +def _parse_instance_creation_timestamp(instance: Instance) -> Optional[datetime]: + try: + return parse_datetime(instance.creation_timestamp) + except ValueError: + logger.warning( + "Failed to parse creation timestamp %r of instance %s", + instance.creation_timestamp, + instance.name, + ) + return None + + +def _get_last_path_segment(path: str) -> str: + """ + Returns the last segment of a scope, location, or resource path, + e.g. `us-central1-a` for `zones/us-central1-a`. + """ + return path.rsplit("/", 1)[-1] + + +def _get_zone_region(zone: str) -> str: + return zone.rsplit("-", 1)[0] + def _is_supported_gcp_instance(instance_name: str, gpu_name: Optional[str]) -> bool: """Check if the instance is supported by dstack.""" diff --git a/src/tests/_internal/core/backends/gcp/test_compute.py b/src/tests/_internal/core/backends/gcp/test_compute.py index 6675d5df5..041b368cf 100644 --- a/src/tests/_internal/core/backends/gcp/test_compute.py +++ b/src/tests/_internal/core/backends/gcp/test_compute.py @@ -1,9 +1,33 @@ +import json +from datetime import datetime, timezone +from typing import Optional +from unittest.mock import Mock + +import google.api_core.exceptions import pytest from google.auth.credentials import AnonymousCredentials +from google.cloud import compute_v1, tpu_v2 from gpuhunt import AcceleratorVendor, CatalogItem from dstack._internal.core.backends.gcp.compute import GCPCompute from dstack._internal.core.backends.gcp.models import GCPConfig, GCPDefaultCreds +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 + +PROJECT_ID = "test-project" +PROVIDER_REF = "abc123def456" +CREATED_AT = datetime(2026, 1, 1, 12, tzinfo=timezone.utc) + + +def _get_compute(mocker, regions: list[str], tpu: bool = False) -> GCPCompute: + mocker.patch( + "dstack._internal.core.backends.gcp.auth.authenticate", + return_value=(AnonymousCredentials(), None), + ) + return GCPCompute( + GCPConfig(project_id=PROJECT_ID, regions=regions, creds=GCPDefaultCreds(), tpu=tpu) + ) class TestGetAllOffersWithAvailability: @@ -53,3 +77,252 @@ def test_returns_tpu_offers_only_if_enabled(self, mocker, tpu_config): if tpu_config.get("tpu"): expected_names.append("v5litepod-8") assert [offer.instance.name for offer in offers] == expected_names + + +def _get_vm( + name: str, provider_ref: Optional[str] = PROVIDER_REF, status: str = "RUNNING" +) -> compute_v1.Instance: + labels = {"owner": "dstack"} + if provider_ref is not None: + labels["dstack_provider_ref"] = provider_ref + return compute_v1.Instance( + name=name, + machine_type=f"https://www.googleapis.com/compute/v1/projects/{PROJECT_ID}/zones/us-central1-a/machineTypes/g2-standard-4", + status=status, + creation_timestamp="2026-01-01T04:00:00.000-08:00", + labels=labels, + ) + + +def _get_vm_pages( + instances_by_zone: dict[str, list[compute_v1.Instance]], + unreachable_zones: tuple[str, ...] = (), +) -> Mock: + items = {} + for zone, instances in instances_by_zone.items(): + items[f"zones/{zone}"] = compute_v1.InstancesScopedList(instances=instances) + for zone in unreachable_zones: + items[f"zones/{zone}"] = compute_v1.InstancesScopedList( + warning=compute_v1.Warning(code="UNREACHABLE") + ) + return Mock(pages=[compute_v1.InstanceAggregatedList(items=items)]) + + +def _get_tpu_node( + node_id: str, zone: str = "us-central1-a", state=tpu_v2.Node.State.READY +) -> tpu_v2.Node: + node = tpu_v2.Node( + name=f"projects/{PROJECT_ID}/locations/{zone}/nodes/{node_id}", + accelerator_type="v5litepod-8", + state=state, + labels={"owner": "dstack", "dstack_provider_ref": PROVIDER_REF}, + ) + node.create_time = CREATED_AT + return node + + +def _get_resource_group(is_tpu: bool = False) -> ListedResourceGroup: + return ListedResourceGroup( + backend=BackendType.GCP, + owner_type=OwnerType.INSTANCE, + provider_ref=PROVIDER_REF, + primary_id="vm-1", + region="us-central1", + created_at=CREATED_AT, + description="instance vm-1", + backend_data=json.dumps({"zone": "us-central1-a", "is_tpu": is_tpu}), + ) + + +class TestListResourceGroups: + def test_lists_labeled_vm(self, mocker): + compute = _get_compute(mocker, regions=["us-central1"]) + aggregated_list = mocker.patch.object( + compute.instances_client, + "aggregated_list", + return_value=_get_vm_pages({"us-central1-a": [_get_vm("vm-1")]}), + ) + + listing = compute.list_resource_groups() + + request = aggregated_list.call_args.kwargs["request"] + assert request.project == PROJECT_ID + assert request.filter == "labels.dstack_provider_ref:*" + assert listing.failures == [] + assert len(listing.resource_groups) == 1 + group = listing.resource_groups[0] + assert group.backend == BackendType.GCP + assert group.owner_type == OwnerType.INSTANCE + assert group.provider_ref == PROVIDER_REF + assert group.primary_id == "vm-1" + assert group.region == "us-central1" + assert group.created_at == CREATED_AT + assert group.description == "instance vm-1 (g2-standard-4, RUNNING)" + assert json.loads(group.backend_data or "") == {"zone": "us-central1-a", "is_tpu": False} + + def test_skips_vms_without_valid_provider_ref(self, mocker): + compute = _get_compute(mocker, regions=["us-central1"]) + vms = [_get_vm("vm-1", provider_ref=None), _get_vm("vm-2", provider_ref="Invalid-Ref")] + mocker.patch.object( + compute.instances_client, + "aggregated_list", + return_value=_get_vm_pages({"us-central1-a": vms}), + ) + assert compute.list_resource_groups().resource_groups == [] + + def test_skips_stopping_vms_and_lists_stopped_ones(self, mocker): + compute = _get_compute(mocker, regions=["us-central1"]) + vms = [_get_vm("vm-1", status="STOPPING"), _get_vm("vm-2", status="TERMINATED")] + mocker.patch.object( + compute.instances_client, + "aggregated_list", + return_value=_get_vm_pages({"us-central1-a": vms}), + ) + groups = compute.list_resource_groups().resource_groups + assert [g.primary_id for g in groups] == ["vm-2"] + + def test_skips_regions_not_in_config(self, mocker): + compute = _get_compute(mocker, regions=["us-central1"]) + mocker.patch.object( + compute.instances_client, + "aggregated_list", + return_value=_get_vm_pages( + {"us-central1-a": [_get_vm("vm-1")], "europe-west4-a": [_get_vm("vm-2")]}, + unreachable_zones=("europe-west4-b",), + ), + ) + listing = compute.list_resource_groups() + assert [g.primary_id for g in listing.resource_groups] == ["vm-1"] + assert listing.failures == [] + + def test_reports_region_with_unreachable_zone_and_lists_other_regions(self, mocker): + compute = _get_compute(mocker, regions=["us-central1", "europe-west4"]) + mocker.patch.object( + compute.instances_client, + "aggregated_list", + return_value=_get_vm_pages( + {"us-central1-a": [_get_vm("vm-1")], "europe-west4-a": [_get_vm("vm-2")]}, + unreachable_zones=("europe-west4-b",), + ), + ) + listing = compute.list_resource_groups() + assert [g.primary_id for g in listing.resource_groups] == ["vm-1"] + assert [(f.backend, f.region) for f in listing.failures] == [ + (BackendType.GCP, "europe-west4") + ] + + def test_raises_if_all_regions_unreachable(self, mocker): + compute = _get_compute(mocker, regions=["us-central1"]) + mocker.patch.object( + compute.instances_client, + "aggregated_list", + return_value=_get_vm_pages({}, unreachable_zones=("us-central1-a",)), + ) + with pytest.raises(ComputeError): + compute.list_resource_groups() + + def test_raises_compute_error_if_not_permitted(self, mocker): + compute = _get_compute(mocker, regions=["us-central1"]) + mocker.patch.object( + compute.instances_client, + "aggregated_list", + side_effect=google.api_core.exceptions.Forbidden("Required permission"), + ) + with pytest.raises(ComputeError, match="compute.instances.list"): + compute.list_resource_groups() + + @pytest.mark.parametrize("tpu", [False, True], ids=["tpu_disabled", "tpu_enabled"]) + def test_lists_tpu_nodes_only_if_enabled(self, mocker, tpu: bool): + compute = _get_compute(mocker, regions=["us-central1"], tpu=tpu) + mocker.patch.object( + compute.instances_client, "aggregated_list", return_value=_get_vm_pages({}) + ) + list_nodes = mocker.patch.object( + compute.tpu_client, + "list_nodes", + return_value=Mock( + pages=[ + tpu_v2.ListNodesResponse( + nodes=[ + _get_tpu_node("tpu-1"), + _get_tpu_node("tpu-2", state=tpu_v2.Node.State.DELETING), + ] + ) + ] + ), + ) + + groups = compute.list_resource_groups().resource_groups + + if not tpu: + list_nodes.assert_not_called() + assert groups == [] + return + assert ( + list_nodes.call_args.kwargs["request"].parent == f"projects/{PROJECT_ID}/locations/-" + ) + assert len(groups) == 1 + group = groups[0] + assert group.primary_id == "tpu-1" + assert group.region == "us-central1" + assert group.created_at == CREATED_AT + assert json.loads(group.backend_data or "") == {"zone": "us-central1-a", "is_tpu": True} + + def test_reports_region_with_unreachable_tpu_location(self, mocker): + compute = _get_compute(mocker, regions=["us-central1", "europe-west4"], tpu=True) + mocker.patch.object( + compute.instances_client, + "aggregated_list", + return_value=_get_vm_pages({"us-central1-a": [_get_vm("vm-1")]}), + ) + mocker.patch.object( + compute.tpu_client, + "list_nodes", + return_value=Mock( + pages=[ + tpu_v2.ListNodesResponse( + nodes=[_get_tpu_node("tpu-1", zone="europe-west4-a")], + unreachable=[f"projects/{PROJECT_ID}/locations/europe-west4-b"], + ) + ] + ), + ) + listing = compute.list_resource_groups() + assert [g.primary_id for g in listing.resource_groups] == ["vm-1"] + assert [f.region for f in listing.failures] == ["europe-west4"] + + +class TestDeleteResourceGroup: + def test_deletes_vm(self, mocker): + compute = _get_compute(mocker, regions=["us-central1"]) + delete = mocker.patch.object(compute.instances_client, "delete") + compute.delete_resource_group(_get_resource_group()) + delete.assert_called_once_with(project=PROJECT_ID, zone="us-central1-a", instance="vm-1") + + def test_deletes_tpu_node(self, mocker): + compute = _get_compute(mocker, regions=["us-central1"], tpu=True) + delete_node = mocker.patch.object(compute.tpu_client, "delete_node") + compute.delete_resource_group(_get_resource_group(is_tpu=True)) + assert ( + delete_node.call_args.kwargs["request"].name + == f"projects/{PROJECT_ID}/locations/us-central1-a/nodes/vm-1" + ) + + def test_tolerates_already_deleted_vm(self, mocker): + compute = _get_compute(mocker, regions=["us-central1"]) + mocker.patch.object( + compute.instances_client, + "delete", + side_effect=google.api_core.exceptions.NotFound("not found"), + ) + compute.delete_resource_group(_get_resource_group()) + + def test_raises_compute_error_on_api_error(self, mocker): + compute = _get_compute(mocker, regions=["us-central1"]) + mocker.patch.object( + compute.instances_client, + "delete", + side_effect=google.api_core.exceptions.Forbidden("Required permission"), + ) + with pytest.raises(ComputeError): + compute.delete_resource_group(_get_resource_group())