diff --git a/runpod/apps/datacenter.py b/runpod/apps/datacenter.py index 1ba61f46..cff8d45c 100644 --- a/runpod/apps/datacenter.py +++ b/runpod/apps/datacenter.py @@ -40,6 +40,22 @@ def from_string(cls, value: str) -> "DataCenter": def all(cls) -> List["DataCenter"]: return list(cls) + @classmethod + def is_known(cls, datacenter_id: str) -> bool: + return datacenter_id in cls._value2member_map_ + + +def datacenter_id(value: str) -> str: + """api id for a datacenter: built-in ids normalize, others pass through. + + the built-in list goes stale as datacenters come online, and the + backend validates ids anyway, so unrecognized ids are only stripped. + """ + try: + return DataCenter.from_string(value).value + except ValueError: + return value.strip() + # datacenters with high cpu serverless stock. storage support is checked # separately; cpu5c/cpu5g are only stocked in EU-RO-1. diff --git a/runpod/apps/placement.py b/runpod/apps/placement.py index e9e99add..beea4f69 100644 --- a/runpod/apps/placement.py +++ b/runpod/apps/placement.py @@ -11,7 +11,7 @@ import logging from typing import Dict, Iterable, List, Optional, Set, Tuple -from .datacenter import DataCenter +from .datacenter import DataCenter, datacenter_id from .errors import AppError from .gpu import GpuGroup from .utils.client import default_client @@ -161,17 +161,21 @@ def candidates(spec, stock: StockMap) -> Set[str]: """ spec.validate() allowed = {dc.value for dc in DataCenter.all()} + # stock is only queried for built-in datacenters, so an unlisted pin + # cannot be ruled out here; the backend decides whether it schedules + unlisted: Set[str] = set() if spec.datacenter: + unlisted = {dc for dc in spec.datacenter if not DataCenter.is_known(dc)} allowed &= set(spec.datacenter) keys = _hardware_keys(spec) if not keys: - return allowed + return allowed | unlisted viable = set() for dc in allowed: if any(stock.score(key, dc) > 0 for key in keys): viable.add(dc) - return viable + return viable | unlisted def _resource_best_score(spec, stock: StockMap, dc: str) -> int: @@ -200,7 +204,7 @@ def solve_placement( per_resource = {spec.name: candidates(spec, stock) for spec in specs} if existing_dc is not None: - existing_dc = DataCenter.from_string(existing_dc).value + existing_dc = datacenter_id(existing_dc) blocked = [name for name, dcs in per_resource.items() if existing_dc not in dcs] if blocked: raise PlacementError( diff --git a/runpod/apps/spec.py b/runpod/apps/spec.py index a6a41373..391471e7 100644 --- a/runpod/apps/spec.py +++ b/runpod/apps/spec.py @@ -5,17 +5,25 @@ and what dev-session provisioning consumes. it holds no live state. """ +import logging from dataclasses import dataclass, field from enum import Enum +from functools import lru_cache from typing import Any, Dict, List, Mapping, Optional, Tuple, Union from .errors import InvalidResourceError -from .datacenter import DataCenter +from .datacenter import DataCenter, datacenter_id from .gpu import GpuGroup, GpuLike, GpuType from .volume import Volume, normalize_mounts, validate_mounts +log = logging.getLogger(__name__) + DEFAULT_WORKERS: Tuple[int, int] = (0, 3) +DATACENTER_IDS_DOCS_URL = ( + "https://docs.runpod.io/api-reference-v2/catalog/list-data-centers" +) + SCALER_TYPES = frozenset({"QUEUE_DELAY", "REQUEST_COUNT"}) DEFAULT_SCALER_VALUE = 4 @@ -147,17 +155,37 @@ def normalize_cuda_version(version: Optional[str]) -> Optional[str]: return normalized +@lru_cache(maxsize=None) +def _warn_unlisted_datacenter(dc_id: str) -> None: + # specs revalidate many times per deploy; warn once per id per process + log.warning( + "%r doesn't look like a valid datacenter id (see %s) but will be used anyway", + dc_id, + DATACENTER_IDS_DOCS_URL, + ) + + def normalize_datacenter( datacenter: Optional[Union[str, List[str]]], ) -> Optional[List[str]]: - """normalize enum values and string variants to api datacenter ids.""" + """normalize enum values and string variants to api datacenter ids. + + ids outside the built-in DataCenter list pass through with a warning; + blank or non-string values raise InvalidResourceError. + """ if datacenter is None: return None values = [datacenter] if isinstance(datacenter, str) else datacenter try: - return [DataCenter.from_string(value).value for value in values] - except (AttributeError, TypeError, ValueError) as exc: + ids = [datacenter_id(value) for value in values] + except (AttributeError, TypeError) as exc: raise InvalidResourceError(f"invalid datacenter: {datacenter!r}") from exc + if not all(ids): + raise InvalidResourceError(f"invalid datacenter: {datacenter!r}") + for dc_id in ids: + if not DataCenter.is_known(dc_id): + _warn_unlisted_datacenter(dc_id) + return ids @dataclass diff --git a/runpod/apps/volume.py b/runpod/apps/volume.py index 7e1989cc..3ae904a3 100644 --- a/runpod/apps/volume.py +++ b/runpod/apps/volume.py @@ -335,10 +335,10 @@ async def resolve( dc = record["dataCenter"] if record is not None else volume.datacenter if record is None: - from .datacenter import DataCenter + from .datacenter import datacenter_id pins = { - DataCenter.from_string(ref.datacenter).value + datacenter_id(ref.datacenter) for ref in [volume] + [ ref diff --git a/tests/test_apps/test_config_params.py b/tests/test_apps/test_config_params.py index f92d4f4f..7ee1ab31 100644 --- a/tests/test_apps/test_config_params.py +++ b/tests/test_apps/test_config_params.py @@ -4,15 +4,20 @@ from runpod.apps import App from runpod.apps.app import _clear_registry +from runpod.apps.datacenter import DataCenter from runpod.apps.dev import _endpoint_input from runpod.apps.errors import InvalidResourceError from runpod.apps.spec import ( + DATACENTER_IDS_DOCS_URL, ResourceKind, ResourceSpec, normalize_cuda_version, + normalize_datacenter, normalize_scaler_type, ) +UNLISTED_DC = "XX-TEST-1" + @pytest.fixture(autouse=True) def clean_registry(): @@ -44,6 +49,55 @@ def test_cuda_version_rejects_unknown(self): normalize_cuda_version("10.0") +class TestDatacenterNormalization: + @pytest.fixture(autouse=True) + def _fresh_warnings(self): + from runpod.apps.spec import _warn_unlisted_datacenter + + _warn_unlisted_datacenter.cache_clear() + yield + _warn_unlisted_datacenter.cache_clear() + + @pytest.mark.parametrize( + "value", [DataCenter.EU_RO_1, "EU-RO-1", "eu_ro_1", " eu-ro-1 ", ["eu_ro_1"]] + ) + def test_known_values_normalize_without_warning(self, value, caplog): + assert normalize_datacenter(value) == ["EU-RO-1"] + assert "valid datacenter id" not in caplog.text + + def test_none_passes(self): + assert normalize_datacenter(None) is None + + def test_unlisted_id_passes_through_with_warning(self, caplog): + assert normalize_datacenter(UNLISTED_DC) == [UNLISTED_DC] + warnings = [r for r in caplog.records if r.levelname == "WARNING"] + assert len(warnings) == 1 + assert UNLISTED_DC in warnings[0].getMessage() + message = warnings[0].getMessage() + assert "doesn't look like a valid datacenter id" in message + assert "will be used anyway" in message + assert DATACENTER_IDS_DOCS_URL in message + + def test_mixed_list_keeps_order(self): + assert normalize_datacenter(["us_ks_2", UNLISTED_DC]) == [ + "US-KS-2", + UNLISTED_DC, + ] + + def test_unlisted_id_warns_once_across_revalidation(self, caplog): + spec = ResourceSpec(kind=ResourceKind.QUEUE, name="q", datacenter=UNLISTED_DC) + spec.validate() + spec.to_manifest() + warnings = [r for r in caplog.records if UNLISTED_DC in r.getMessage()] + assert len(warnings) == 1 + assert spec.datacenter == [UNLISTED_DC] + + @pytest.mark.parametrize("value", ["", " ", [""], ["EU-RO-1", " "], [None], 42]) + def test_blank_or_non_string_rejected(self, value): + with pytest.raises(InvalidResourceError, match="invalid datacenter"): + normalize_datacenter(value) + + class TestSpecValidation: @pytest.mark.parametrize( "config", diff --git a/tests/test_apps/test_deploy.py b/tests/test_apps/test_deploy.py index a29d3e40..fe391efe 100644 --- a/tests/test_apps/test_deploy.py +++ b/tests/test_apps/test_deploy.py @@ -132,6 +132,20 @@ def value(self): assert env["PORT_HEALTH"] == "80" + @pytest.mark.parametrize( + "hardware", [{"gpu": "ADA_24"}, {"cpu": "cpu3c-1-2"}, {"cpu": "cpu5c-2-4"}] + ) + def test_unlisted_datacenter_reaches_locations(self, hardware): + app = App("dc-app") + + @app.queue(name="q", datacenter="XX-TEST-1", **hardware) + def q(): + pass + + payload = _deployed_endpoint_input(app, q.spec, "env-1", "build-1", "3.12") + assert payload["locations"] == "XX-TEST-1" + + class TestPackaging: def test_tarball_contains_source_and_manifest(self, tmp_path): _write_project(tmp_path) diff --git a/tests/test_apps/test_placement.py b/tests/test_apps/test_placement.py index 99aef01d..bccd11a7 100644 --- a/tests/test_apps/test_placement.py +++ b/tests/test_apps/test_placement.py @@ -79,6 +79,20 @@ def test_datacenter_variants_match_api_ids(self, pin): assert candidates(spec, stock) == {"EU-RO-1"} assert spec.to_manifest()["locations"] == "EU-RO-1" + def test_unlisted_pin_stays_candidate_without_stock_data(self): + # the sdk only queries stock for its built-in list, so an unlisted + # pin cannot be ruled out client-side; the backend decides + stock = _stock(gpu={("NVIDIA GeForce RTX 4090", "EU-RO-1"): 3}) + spec = _spec(gpu=["NVIDIA GeForce RTX 4090"], datacenter=["XX-TEST-1"]) + assert candidates(spec, stock) == {"XX-TEST-1"} + + def test_unlisted_pin_alongside_known_pin(self): + stock = _stock(gpu={("NVIDIA GeForce RTX 4090", "EU-RO-1"): 0}) + spec = _spec( + gpu=["NVIDIA GeForce RTX 4090"], datacenter=["EU-RO-1", "XX-TEST-1"] + ) + assert candidates(spec, stock) == {"XX-TEST-1"} + def test_mutated_datacenter_pin_is_normalized(self): spec = _spec(cpu="cpu5c-2-4") spec.datacenter = [DataCenter.EU_RO_1] @@ -145,6 +159,36 @@ def test_existing_dc_is_hard_constraint(self): ) assert dc == "EU-RO-1" + def test_existing_unlisted_dc_with_matching_pin(self): + dc = solve_placement( + [_spec("train", gpu=["NVIDIA H200"], datacenter="XX-TEST-1")], + _stock(), + volume_name="models", + existing_dc="XX-TEST-1", + volume_datacenters=set(), + ) + assert dc == "XX-TEST-1" + + def test_existing_unlisted_dc_without_pin_is_placement_error(self): + with pytest.raises(PlacementError, match="lives in XX-TEST-1"): + solve_placement( + [_spec("train", gpu=["NVIDIA H200"])], + _stock(gpu={("NVIDIA H200", "EU-RO-1"): 3}), + volume_name="models", + existing_dc="XX-TEST-1", + volume_datacenters=set(), + ) + + def test_new_volume_in_unlisted_dc(self): + dc = solve_placement( + [_spec("train", gpu=["NVIDIA H200"], datacenter="XX-TEST-1")], + _stock(), + volume_name="models", + volume_dc="XX-TEST-1", + volume_datacenters={"EU-RO-1", "XX-TEST-1"}, + ) + assert dc == "XX-TEST-1" + def test_existing_dc_unschedulable_errors(self): stock = _stock(gpu={("NVIDIA H200", "EU-RO-1"): 3}) with pytest.raises(PlacementError, match="lives in US-KS-2"): diff --git a/tests/test_apps/test_volume.py b/tests/test_apps/test_volume.py index 0093ce2b..6306eb98 100644 --- a/tests/test_apps/test_volume.py +++ b/tests/test_apps/test_volume.py @@ -317,6 +317,22 @@ def test_creation_without_app_requires_explicit_placement(self): ) assert resolved == {"id": "nv-new", "dataCenterId": "EU-RO-1"} + async def test_unlisted_volume_pin_uses_catalog(self): + api = _api( + created={"id": "nv-new", "name": "models", "size": 50, "dataCenter": "XX-TEST-1"} + ) + api.network_volume_datacenters.return_value = {"EU-RO-1", "XX-TEST-1"} + spec = ResourceSpec( + kind=ResourceKind.TASK, name="r", gpu=None, datacenter="XX-TEST-1" + ) + resolved = await VolumeResolver(api).resolve( + NetworkVolume("models", datacenter="XX-TEST-1"), [spec] + ) + assert resolved == {"id": "nv-new", "dataCenterId": "XX-TEST-1"} + api.create_network_volume.assert_awaited_once_with( + name="models", size=50, data_center_id="XX-TEST-1" + ) + async def test_unsupported_explicit_pin_never_relocates(self): api = _api() api.network_volume_datacenters.return_value = {"EU-RO-1"}