From ce3a87bd60ea3e07d3ab959887591765318657d9 Mon Sep 17 00:00:00 2001 From: Andrey Cheptsov Date: Wed, 14 Oct 2026 20:02:54 +0200 Subject: [PATCH 1/2] Show quota and balance provisioning failures --- .../cli/services/configurators/run.py | 7 +- src/dstack/_internal/cli/utils/fleet.py | 11 +- src/dstack/_internal/core/models/instances.py | 2 + src/dstack/_internal/core/models/runs.py | 8 + .../background/pipeline_tasks/fleets.py | 11 +- .../instances/cloud_provisioning.py | 22 +- .../pipeline_tasks/jobs_submitted.py | 22 +- .../background/pipeline_tasks/runs/common.py | 11 +- .../server/services/backends/__init__.py | 20 +- .../_internal/server/services/fleets.py | 17 +- .../server/services/jobs/__init__.py | 8 + .../_internal/server/services/offers.py | 213 +++++++++---- .../server/services/runs/__init__.py | 6 +- .../_internal/server/services/runs/plan.py | 6 +- src/tests/_internal/cli/utils/test_fleet.py | 18 ++ src/tests/_internal/cli/utils/test_run.py | 48 +++ src/tests/_internal/core/models/test_runs.py | 15 + .../background/pipeline_tasks/test_fleets.py | 16 +- .../test_instances/test_cloud_provisioning.py | 57 +++- .../pipeline_tasks/test_runs/test_active.py | 24 +- .../pipeline_tasks/test_runs/test_pending.py | 16 +- .../pipeline_tasks/test_submitted_jobs.py | 169 +++++++++- .../server/services/jobs/test_jobs.py | 32 +- .../server/services/runs/test_plan.py | 3 +- .../server/services/runs/test_runs.py | 11 +- .../_internal/server/services/test_offers.py | 298 +++++++++++++++++- 26 files changed, 934 insertions(+), 137 deletions(-) diff --git a/src/dstack/_internal/cli/services/configurators/run.py b/src/dstack/_internal/cli/services/configurators/run.py index d3273ba9c6..7429737a9a 100644 --- a/src/dstack/_internal/cli/services/configurators/run.py +++ b/src/dstack/_internal/cli/services/configurators/run.py @@ -894,7 +894,12 @@ def print_finished_message(run: Run): if ( termination_reason # A run that never started has no runner logs to read. - and termination_reason != JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY.value + and termination_reason + not in ( + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY.value, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA.value, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE.value, + ) ): console.print(f"Check [code]dstack logs -d {run.name}[/code] for more details.") diff --git a/src/dstack/_internal/cli/utils/fleet.py b/src/dstack/_internal/cli/utils/fleet.py index 811db7db9f..6963e493be 100644 --- a/src/dstack/_internal/cli/utils/fleet.py +++ b/src/dstack/_internal/cli/utils/fleet.py @@ -11,7 +11,11 @@ from dstack._internal.core.models.backends.base import BackendType from dstack._internal.core.models.fleets import Fleet, FleetNodesSpec, FleetStatus from dstack._internal.core.models.health import HealthStatus -from dstack._internal.core.models.instances import Instance, InstanceStatus +from dstack._internal.core.models.instances import ( + Instance, + InstanceStatus, + InstanceTerminationReason, +) from dstack._internal.core.models.resources import GPUSpec, ResourcesSpec from dstack._internal.utils.common import DateFormatter, pretty_date @@ -166,6 +170,11 @@ def get_fleets_table( if instance.status == InstanceStatus.TERMINATED and instance.termination_reason: instance_row["ERROR"] = instance.termination_reason + if ( + instance.termination_reason == InstanceTerminationReason.NO_OFFERS + and instance.termination_reason_message in ("No quota", "No balance") + ): + instance_row["ERROR"] = instance.termination_reason_message.lower() # No row-level dimming: only the columns styled above are dimmed on instance rows add_row_from_dict(table, instance_row) diff --git a/src/dstack/_internal/core/models/instances.py b/src/dstack/_internal/core/models/instances.py index 68e5c11028..e0fdfa227c 100644 --- a/src/dstack/_internal/core/models/instances.py +++ b/src/dstack/_internal/core/models/instances.py @@ -246,6 +246,8 @@ class InstanceTerminationReason(str, Enum): JOB_FINISHED = "job_finished" UNREACHABLE = "unreachable" NO_OFFERS = "no_offers" + # Add read support before emission so server replicas can be upgraded gradually. + NO_QUOTA = "no_quota" MASTER_FAILED = "master_failed" MAX_INSTANCES_LIMIT = "max_instances_limit" FLEET_SPEC_MISMATCH = "fleet_spec_mismatch" diff --git a/src/dstack/_internal/core/models/runs.py b/src/dstack/_internal/core/models/runs.py index a120390079..032e6bea65 100644 --- a/src/dstack/_internal/core/models/runs.py +++ b/src/dstack/_internal/core/models/runs.py @@ -138,6 +138,9 @@ def to_error(self) -> Optional[str]: class JobTerminationReason(str, Enum): # Set by the server FAILED_TO_START_DUE_TO_NO_CAPACITY = "failed_to_start_due_to_no_capacity" + # Add read support before emission so server replicas can be upgraded gradually. + FAILED_TO_START_DUE_TO_NO_QUOTA = "failed_to_start_due_to_no_quota" + FAILED_TO_START_DUE_TO_NO_BALANCE = "failed_to_start_due_to_no_balance" INTERRUPTED_BY_NO_CAPACITY = "interrupted_by_no_capacity" INSTANCE_UNREACHABLE = "instance_unreachable" INSTANCE_ACCESS_REVOKED = "instance_access_revoked" @@ -163,6 +166,8 @@ class JobTerminationReason(str, Enum): def to_status(self) -> JobStatus: mapping = { self.FAILED_TO_START_DUE_TO_NO_CAPACITY: JobStatus.FAILED, + self.FAILED_TO_START_DUE_TO_NO_QUOTA: JobStatus.FAILED, + self.FAILED_TO_START_DUE_TO_NO_BALANCE: JobStatus.FAILED, self.INTERRUPTED_BY_NO_CAPACITY: JobStatus.FAILED, self.INSTANCE_UNREACHABLE: JobStatus.FAILED, self.INSTANCE_ACCESS_REVOKED: JobStatus.FAILED, @@ -194,6 +199,9 @@ def to_retry_event(self) -> Optional[RetryEvent]: """ mapping = { self.FAILED_TO_START_DUE_TO_NO_CAPACITY: RetryEvent.NO_CAPACITY, + # Preserve existing no-capacity retry configurations for blocked offers. + self.FAILED_TO_START_DUE_TO_NO_QUOTA: RetryEvent.NO_CAPACITY, + self.FAILED_TO_START_DUE_TO_NO_BALANCE: RetryEvent.NO_CAPACITY, self.INTERRUPTED_BY_NO_CAPACITY: RetryEvent.INTERRUPTION, } default = RetryEvent.ERROR if self.to_status() == JobStatus.FAILED else None diff --git a/src/dstack/_internal/server/background/pipeline_tasks/fleets.py b/src/dstack/_internal/server/background/pipeline_tasks/fleets.py index 4904e7e3a2..ff285ae5eb 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/fleets.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/fleets.py @@ -819,8 +819,8 @@ def _set_fail_instances_on_master_bootstrap_failure( instance_id_to_update_map: dict[uuid.UUID, _InstanceUpdateMap], ) -> None: """ - Terminates instances with MASTER_FAILED if the master dies with NO_OFFERS in a cluster with node.min == 0. - This is needed to avoid master re-election loop and fail fast. + Terminates unprovisioned siblings when the cluster master has no usable offers. + For nodes.min == 0, this prevents a master re-election loop. """ fleet_spec = get_fleet_spec(fleet_model) if ( @@ -841,7 +841,12 @@ def _set_fail_instances_on_master_bootstrap_failure( if ( current_master_instance_model.status != InstanceStatus.TERMINATED - or current_master_instance_model.termination_reason != InstanceTerminationReason.NO_OFFERS + or current_master_instance_model.termination_reason + not in ( + InstanceTerminationReason.NO_OFFERS, + InstanceTerminationReason.NO_QUOTA, + InstanceTerminationReason.NO_BALANCE, + ) ): return diff --git a/src/dstack/_internal/server/background/pipeline_tasks/instances/cloud_provisioning.py b/src/dstack/_internal/server/background/pipeline_tasks/instances/cloud_provisioning.py index bd869b8325..de8e605c37 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/instances/cloud_provisioning.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/instances/cloud_provisioning.py @@ -23,6 +23,7 @@ SkipOffer, ) from dstack._internal.core.models.instances import ( + InstanceAvailability, InstanceOfferWithAvailability, InstanceStatus, InstanceTerminationReason, @@ -116,7 +117,24 @@ async def create_cloud_instance(instance_model: InstanceModel) -> ProcessResult: include_only_create_instance_supported_backends=True, ) - offers_iter = iter(offers) + available_offers = offers.offers + for availability, message in ( + (InstanceAvailability.NO_QUOTA, "No quota"), + (InstanceAvailability.NO_BALANCE, "No balance"), + ): + if offers.unavailable_reason == availability: + set_status_update( + update_map=result.instance_update_map, + instance_model=instance_model, + new_status=InstanceStatus.TERMINATED, + termination_reason=InstanceTerminationReason.NO_OFFERS, + # TODO: Store NO_QUOTA/NO_BALANCE once the minimum supported rolling-upgrade + # source version handles them on every replica. + termination_reason_message=message, + ) + return result + + offers_iter = iter(available_offers) offers_tried = 0 # Limit number of offers tried to prevent long-running processing in case all offers fail. while offers_tried < server_settings.MAX_OFFERS_TRIED: @@ -241,7 +259,7 @@ async def create_cloud_instance(instance_model: InstanceModel) -> ProcessResult: instance_model=instance_model, new_status=InstanceStatus.TERMINATED, termination_reason=InstanceTerminationReason.NO_OFFERS, - termination_reason_message="All offers failed" if offers else "No offers found", + termination_reason_message="All offers failed" if available_offers else "No offers found", ) return result diff --git a/src/dstack/_internal/server/background/pipeline_tasks/jobs_submitted.py b/src/dstack/_internal/server/background/pipeline_tasks/jobs_submitted.py index 1ab7875d04..041574799d 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/jobs_submitted.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/jobs_submitted.py @@ -31,6 +31,7 @@ InstanceGroupPlacement, ) from dstack._internal.core.models.instances import ( + InstanceAvailability, InstanceOfferWithAvailability, InstanceStatus, ) @@ -485,7 +486,7 @@ class _NewCapacityAttempts: class _FailedNewCapacityProvisioning: placement_group_cleanup: Optional[_PlacementGroupCleanup] message: Optional[str] = None - """Why the job could not be provisioned. `None` if the offers were never tried.""" + """Why the job could not be provisioned.""" @dataclass @@ -2420,8 +2421,21 @@ async def _provision_new_capacity( instance_mounts=check_run_spec_requires_instance_mounts(run.run_spec), placement_group=placement_group_model_to_placement_group_optional(placement_group_model), ) + available_offers = offers.offers + for availability, message in ( + (InstanceAvailability.NO_QUOTA, "No quota"), + (InstanceAvailability.NO_BALANCE, "No balance"), + ): + if offers.unavailable_reason == availability: + return _FailedNewCapacityProvisioning( + placement_group_cleanup=None, + # TODO: Store FAILED_TO_START_DUE_TO_NO_QUOTA/NO_BALANCE once the minimum + # supported rolling-upgrade source version handles them on every replica. + message=message, + ) + extra_authorized_keys = get_extra_authorized_keys(run.run_spec) - offers_iter = iter(offers) + offers_iter = iter(available_offers) offers_tried = 0 offers_taken = 0 skip_reasons: list[str] = [] @@ -2573,12 +2587,12 @@ async def _provision_new_capacity( message=_get_new_capacity_failure_message( fleet_name=fleet_model.name, attempts=_NewCapacityAttempts( - total=len(offers), + total=len(available_offers), tried=offers_tried, skip_reasons=skip_reasons, errors=offer_errors, # Offers are left only if the attempt limit, not the offer list, ended the loop. - limit_reached=offers_taken < len(offers), + limit_reached=offers_taken < len(available_offers), ), ), placement_group_cleanup=_build_placement_group_cleanup( diff --git a/src/dstack/_internal/server/background/pipeline_tasks/runs/common.py b/src/dstack/_internal/server/background/pipeline_tasks/runs/common.py index 03c528c9c1..ff30950808 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/runs/common.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/runs/common.py @@ -127,7 +127,7 @@ async def delete_superseded_no_capacity_job_submissions( new_job_models: list[JobModel], ) -> None: """ - Delete previous job submissions that failed to start due to no capacity without + Delete previous job submissions that failed to start due to capacity, quota, or balance without ever provisioning. Such submissions are created each time a run is retrying on no capacity and carry no information beyond the events. The direct predecessor of each new submission is kept so that the run's @@ -143,8 +143,13 @@ async def delete_superseded_no_capacity_job_submissions( JobModel.job_num == job_model.job_num, JobModel.submission_num < job_model.submission_num - 1, JobModel.status.in_(JobStatus.finished_statuses()), - JobModel.termination_reason - == JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, + JobModel.termination_reason.in_( + [ + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE, + ] + ), JobModel.job_provisioning_data.is_(None), ) ) diff --git a/src/dstack/_internal/server/services/backends/__init__.py b/src/dstack/_internal/server/services/backends/__init__.py index 8472375b7a..f15845b82c 100644 --- a/src/dstack/_internal/server/services/backends/__init__.py +++ b/src/dstack/_internal/server/services/backends/__init__.py @@ -40,7 +40,6 @@ from dstack._internal.core.models.runs import Requirements from dstack._internal.server import settings from dstack._internal.server.models import BackendModel, DecryptedString, ProjectModel -from dstack._internal.server.services.offers import merge_offer_iterables from dstack._internal.utils.common import run_async from dstack._internal.utils.logging import get_logger @@ -472,26 +471,17 @@ async def get_backend_offers( requirements: Requirements, full_offers: bool, unallocated_resources: bool, - exclude_not_available: bool = False, -) -> Iterable[Tuple[Backend, InstanceOfferWithAvailability]]: +) -> list[tuple[Backend, Iterable[InstanceOfferWithAvailability]]]: """ - Returns a lazy iterable of backend offers satisfying `requirements` sorted by price. + Return each backend's lazy offers satisfying `requirements`. """ - def get_filtered_offers_with_backends( - backend: Backend, - offers: Iterable[InstanceOfferWithAvailability], - ) -> Iterator[Tuple[Backend, InstanceOfferWithAvailability]]: - for offer in offers: - if not exclude_not_available or offer.availability.is_available(): - yield (backend, offer) - logger.debug("Requesting instance offers from backends: %s", [b.TYPE.value for b in backends]) tasks = [ run_async(get_offers_tracked, backend, requirements, full_offers, unallocated_resources) for backend in backends ] - offers_by_backend: list[Iterable[tuple[Backend, InstanceOfferWithAvailability]]] = [] + offers_by_backend: list[tuple[Backend, Iterable[InstanceOfferWithAvailability]]] = [] for backend, result in zip(backends, await asyncio.gather(*tasks, return_exceptions=True)): if isinstance(result, BackendError): logger.warning( @@ -507,8 +497,8 @@ def get_filtered_offers_with_backends( exc_info=result, ) continue - offers_by_backend.append(get_filtered_offers_with_backends(backend, result)) - return merge_offer_iterables(*offers_by_backend) + offers_by_backend.append((backend, result)) + return offers_by_backend def check_backend_type_available(backend_type: BackendType): diff --git a/src/dstack/_internal/server/services/fleets.py b/src/dstack/_internal/server/services/fleets.py index ac677b4948..6bdaa9b56f 100644 --- a/src/dstack/_internal/server/services/fleets.py +++ b/src/dstack/_internal/server/services/fleets.py @@ -3,13 +3,12 @@ from collections.abc import Callable from datetime import datetime from functools import wraps -from typing import List, Literal, Optional, Tuple, TypeVar, Union +from typing import List, Literal, Optional, TypeVar, Union from sqlalchemy import and_, exists, false, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import aliased, joinedload, selectinload -from dstack._internal.core.backends.base.backend import Backend from dstack._internal.core.backends.features import BACKENDS_WITH_CREATE_INSTANCE_SUPPORT from dstack._internal.core.errors import ( ForbiddenError, @@ -32,7 +31,6 @@ SSHParams, ) from dstack._internal.core.models.instances import ( - InstanceOfferWithAvailability, InstanceStatus, InstanceTerminationReason, SSHConnectionParams, @@ -470,7 +468,7 @@ async def get_plan( include_only_create_instance_supported_backends ), ) - offers = [offer for _, offer in offers_with_backends] + offers = [offer for _, offer in offers_with_backends.offers] _remove_fleet_spec_sensitive_info(effective_spec) if current_fleet is not None: @@ -501,7 +499,7 @@ async def get_fleet_offers( master_job_provisioning_data: Optional[JobProvisioningData] = None, infer_master_job_provisioning_data_from_fleet_instances: bool = True, include_only_create_instance_supported_backends: bool = True, -) -> List[Tuple[Backend, InstanceOfferWithAvailability]]: +) -> offers_services.Offers: """ Return offers for fleet planning and provisioning. @@ -536,13 +534,10 @@ async def get_fleet_offers( master_job_provisioning_data=master_job_provisioning_data, placement_group=placement_group, blocks=blocks, + supported_backend_types=BACKENDS_WITH_CREATE_INSTANCE_SUPPORT + if include_only_create_instance_supported_backends + else None, ) - if include_only_create_instance_supported_backends: - offers = [ - (backend, offer) - for backend, offer in offers - if offer.backend in BACKENDS_WITH_CREATE_INSTANCE_SUPPORT - ] return offers diff --git a/src/dstack/_internal/server/services/jobs/__init__.py b/src/dstack/_internal/server/services/jobs/__init__.py index 0326be9aca..5a15e8403f 100644 --- a/src/dstack/_internal/server/services/jobs/__init__.py +++ b/src/dstack/_internal/server/services/jobs/__init__.py @@ -625,7 +625,15 @@ def _get_job_status_message(job_model: JobModel) -> str: and "No matching fleet found" in job_model.termination_reason_message ): return "no fleets" + if job_model.termination_reason_message in ("No quota", "No balance"): + return job_model.termination_reason_message.lower() return "no capacity" + elif job_model.termination_reason == JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA: + return "no quota" + elif ( + job_model.termination_reason == JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE + ): + return "no balance" elif job_model.termination_reason == JobTerminationReason.INTERRUPTED_BY_NO_CAPACITY: return "interrupted" else: diff --git a/src/dstack/_internal/server/services/offers.py b/src/dstack/_internal/server/services/offers.py index eb826e03ee..5dea8d665a 100644 --- a/src/dstack/_internal/server/services/offers.py +++ b/src/dstack/_internal/server/services/offers.py @@ -1,6 +1,7 @@ import heapq import itertools from collections.abc import Container, Iterable, Iterator +from dataclasses import dataclass from typing import List, Literal, Optional, Tuple, TypeVar, Union import gpuhunt @@ -15,6 +16,7 @@ ) from dstack._internal.core.models.backends.base import BackendType from dstack._internal.core.models.instances import ( + InstanceAvailability, InstanceOfferWithAvailability, InstanceType, Resources, @@ -28,11 +30,18 @@ from dstack._internal.utils.common import run_async +@dataclass +class Offers: + offers: list[tuple[Backend, InstanceOfferWithAvailability]] + unavailable_reason: Optional[InstanceAvailability] = None + """Common reason for excluded matching offers if none are available; None otherwise.""" + + async def get_offers_by_requirements( project: ProjectModel, profile: Profile, requirements: Requirements, - exclude_not_available=False, + exclude_not_available: bool = False, multinode: bool = False, master_job_provisioning_data: Optional[JobProvisioningData] = None, volumes: Optional[List[List[Volume]]] = None, @@ -43,7 +52,8 @@ async def get_offers_by_requirements( max_offers: Optional[int] = None, full_offers: bool = False, unallocated_resources: bool = False, -) -> List[Tuple[Backend, InstanceOfferWithAvailability]]: + supported_backend_types: Optional[Container[BackendType]] = None, +) -> Offers: backends: List[Backend] = await backends_services.get_project_backends(project=project) backend_types: Optional[list[BackendType]] = profile.backends @@ -99,43 +109,82 @@ async def get_offers_by_requirements( if backend_types is not None: backends = [b for b in backends if b.TYPE in backend_types or b.TYPE == BackendType.DSTACK] - offers = await backends_services.get_backend_offers( + offer_filter = _OfferFilter( + backend_types=backend_types, + regions=None if regions is None else [r.lower() for r in regions], + availability_zones=None + if availability_zones is None + else [z.lower() for z in availability_zones], + instance_types=None if instance_types is None else [i.lower() for i in instance_types], + placement_group=placement_group, + volumes_locations=volumes_locations, + supported_backend_types=supported_backend_types, + ) + + offers_by_backend = await backends_services.get_backend_offers( backends=backends, requirements=requirements, full_offers=full_offers, unallocated_resources=unallocated_resources, - exclude_not_available=exclude_not_available, ) - - offers = _filter_offers( - offers=offers, - # Double filtering by backends if backend returns offers for other backend. - backend_types=backend_types, - regions=regions, - availability_zones=availability_zones, - instance_types=instance_types, - placement_group=placement_group, - volumes_locations=volumes_locations, + return await run_async( + _select_offers, offers_by_backend, offer_filter, blocks, exclude_not_available, max_offers ) + +def _select_offers( + offers_by_backend: list[tuple[Backend, Iterable[InstanceOfferWithAvailability]]], + offer_filter: "_OfferFilter", + blocks: Union[int, Literal["auto"]], + exclude_not_available: bool, + max_offers: Optional[int], +) -> Offers: + unavailable: set[InstanceAvailability] = set() + + def available_offers( + backend: Backend, backend_offers: Iterable[InstanceOfferWithAvailability] + ) -> Iterator[tuple[Backend, InstanceOfferWithAvailability]]: + for offer in backend_offers: + if offer.availability.is_available(): + yield backend, offer + elif ( + offer.availability not in unavailable + and offer_filter.matches(backend, offer) + and (blocks == 1 or _get_offer_blocks(offer, blocks)[0]) + ): + unavailable.add(offer.availability) + + iterables = [ + available_offers(backend, backend_offers) + if exclude_not_available + else _with_backend(backend, backend_offers) + for backend, backend_offers in offers_by_backend + ] + + # Availability must be filtered before merging to preserve backend-specific order. + offers = _filter_offers(merge_offer_iterables(*iterables), offer_filter) + if blocks != 1: offers = _get_shareable_offers(offers, blocks) if max_offers is not None: offers = itertools.islice(offers, max_offers) - # Sorting consumes the lazy offers chain, which is CPU-bound, - # so it must not run on the event loop. - return await run_async(_sort_offers_by_availability, offers) - - -def _sort_offers_by_availability( - offers: Iterable[Tuple[Backend, InstanceOfferWithAvailability]], -) -> List[Tuple[Backend, InstanceOfferWithAvailability]]: # Put NOT_AVAILABLE and NO_QUOTA offers at the end. # We have to do this after taking max_offers to avoid processing all offers # if all/most offers are unavailable. - return sorted(offers, key=lambda i: not i[1].availability.is_available()) + selected = sorted(offers, key=lambda i: not i[1].availability.is_available()) + unavailable_reason = ( + next(iter(unavailable)) if not selected and len(unavailable) == 1 else None + ) + return Offers(offers=selected, unavailable_reason=unavailable_reason) + + +def _with_backend( + backend: Backend, offers: Iterable[InstanceOfferWithAvailability] +) -> Iterator[tuple[Backend, InstanceOfferWithAvailability]]: + for offer in offers: + yield backend, offer T = TypeVar("T") @@ -216,56 +265,81 @@ def get_instance_offer_with_restricted_az( return instance_offer +@dataclass +class _OfferFilter: + backend_types: Optional[List[BackendType]] + regions: Optional[List[str]] + availability_zones: Optional[List[str]] + instance_types: Optional[List[str]] + placement_group: Optional[PlacementGroup] + volumes_locations: Optional[Container[tuple[BackendType, Optional[str]]]] + supported_backend_types: Optional[Container[BackendType]] + + def matches( + self, + backend: Backend, + offer: InstanceOfferWithAvailability, + check_availability_zones: bool = True, + ) -> bool: + # Backends such as dstack can return offers for other backend types. + if self.backend_types is not None and offer.backend not in self.backend_types: + return False + if ( + self.supported_backend_types is not None + and offer.backend not in self.supported_backend_types + ): + return False + if self.regions is not None and offer.region.lower() not in self.regions: + return False + if ( + self.instance_types is not None + and offer.instance.name.lower() not in self.instance_types + ): + return False + if self.placement_group is not None: + compute = backend.compute() + if not isinstance( + compute, ComputeWithPlacementGroupSupport + ) or not compute.is_suitable_placement_group(self.placement_group, offer): + return False + if ( + check_availability_zones + and self.availability_zones is not None + and ( + offer.availability_zones is None + or not any(z.lower() in self.availability_zones for z in offer.availability_zones) + ) + ): + return False + # Offer is futher filtered against volumes AZs in Compute implementation, see + # ComputeWithCreateInstanceSupport._restrict_instance_offer_az_to_volumes_az() + if ( + self.volumes_locations is not None + and (offer.backend, offer.region.lower()) not in self.volumes_locations + and (offer.backend, None) not in self.volumes_locations + ): + return False + return True + + def _filter_offers( offers: Iterable[Tuple[Backend, InstanceOfferWithAvailability]], - backend_types: Optional[List[BackendType]] = None, - regions: Optional[List[str]] = None, - availability_zones: Optional[List[str]] = None, - instance_types: Optional[List[str]] = None, - placement_group: Optional[PlacementGroup] = None, - volumes_locations: Optional[Container[tuple[BackendType, Optional[str]]]] = None, + offer_filter: _OfferFilter, ) -> Iterator[Tuple[Backend, InstanceOfferWithAvailability]]: - """ - Yields filtered offers. May return modified offers to match the filters. - """ - if regions is not None: - regions = [r.lower() for r in regions] - if instance_types is not None: - instance_types = [i.lower() for i in instance_types] - if availability_zones is not None: - availability_zones = [z.lower() for z in availability_zones] - for b, offer in offers: - if backend_types is not None and offer.backend not in backend_types: + # Check AZs once while preparing the restricted offer below. + if not offer_filter.matches(b, offer, check_availability_zones=False): continue - if regions is not None and offer.region.lower() not in regions: - continue - if instance_types is not None and offer.instance.name.lower() not in instance_types: - continue - if placement_group is not None: - compute = b.compute() - if not isinstance( - compute, ComputeWithPlacementGroupSupport - ) or not compute.is_suitable_placement_group(placement_group, offer): - continue - if availability_zones is not None: + if offer_filter.availability_zones is not None: if offer.availability_zones is None: continue new_offer = offer.model_copy() new_offer.availability_zones = [ - z for z in offer.availability_zones if z.lower() in availability_zones + z for z in offer.availability_zones if z.lower() in offer_filter.availability_zones ] if not new_offer.availability_zones: continue offer = new_offer - # Offer is futher filtered against volumes AZs in Compute implementation, see - # ComputeWithCreateInstanceSupport._restrict_instance_offer_az_to_volumes_az() - if ( - volumes_locations is not None - and (offer.backend, offer.region.lower()) not in volumes_locations - and (offer.backend, None) not in volumes_locations - ): - continue yield (b, offer) @@ -277,15 +351,20 @@ def _get_shareable_offers( Yields offers that can be shared with `total_blocks` set. """ for backend, offer in offers: - resources = offer.instance.resources - cpu_count = resources.cpus - gpu_count = len(resources.gpus) - if gpu_count > 0 and resources.gpus[0].vendor == gpuhunt.AcceleratorVendor.GOOGLE: - # TPUs cannot be shared - gpu_count = 1 - divisible, total_blocks = is_divisible_into_blocks(cpu_count, gpu_count, blocks) + divisible, total_blocks = _get_offer_blocks(offer, blocks) if not divisible: continue new_offer = offer.model_copy() new_offer.total_blocks = total_blocks yield (backend, new_offer) + + +def _get_offer_blocks( + offer: InstanceOfferWithAvailability, blocks: Union[int, Literal["auto"]] +) -> tuple[bool, int]: + resources = offer.instance.resources + gpu_count = len(resources.gpus) + if gpu_count > 0 and resources.gpus[0].vendor == gpuhunt.AcceleratorVendor.GOOGLE: + # TPUs cannot be shared + gpu_count = 1 + return is_divisible_into_blocks(resources.cpus, gpu_count, blocks) diff --git a/src/dstack/_internal/server/services/runs/__init__.py b/src/dstack/_internal/server/services/runs/__init__.py index b6ff199641..336d11fb38 100644 --- a/src/dstack/_internal/server/services/runs/__init__.py +++ b/src/dstack/_internal/server/services/runs/__init__.py @@ -1132,7 +1132,11 @@ def _get_run_status_message(run_model: RunModel, job_models: List[JobModel]) -> last_job_termination_reason = _get_last_job_termination_reason(job_models) if ( last_job_termination_reason - == JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY + in ( + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE, + ) and RetryEvent.NO_CAPACITY in retry_on_events ): # TODO: Show `retrying` for other retry events diff --git a/src/dstack/_internal/server/services/runs/plan.py b/src/dstack/_internal/server/services/runs/plan.py index c5c175c57b..4b5e3e5895 100644 --- a/src/dstack/_internal/server/services/runs/plan.py +++ b/src/dstack/_internal/server/services/runs/plan.py @@ -794,7 +794,7 @@ async def _get_backend_offers_in_fleet( # Handle multinode for old jobs that don't have requirements.multinode set. # TODO: Drop multinode param. multinode = requirements.multinode or is_multinode_job(job) - backend_offers = await get_offers_by_requirements( + offers_result = await get_offers_by_requirements( project=project, profile=profile, requirements=requirements, @@ -807,6 +807,7 @@ async def _get_backend_offers_in_fleet( full_offers=full_offers, unallocated_resources=unallocated_resources, ) + backend_offers = offers_result.offers return backend_offers @@ -874,7 +875,7 @@ async def get_non_fleet_offers( if skip_backend_offers: backend_offers = [] else: - backend_offers = await get_offers_by_requirements( + offers_result = await get_offers_by_requirements( project=project, profile=run_spec.merged_profile, requirements=job.job_spec.requirements, @@ -886,6 +887,7 @@ async def get_non_fleet_offers( full_offers=full_offers, unallocated_resources=unallocated_resources, ) + backend_offers = offers_result.offers return instance_offers, backend_offers diff --git a/src/tests/_internal/cli/utils/test_fleet.py b/src/tests/_internal/cli/utils/test_fleet.py index f8a6254758..06e534c4f1 100644 --- a/src/tests/_internal/cli/utils/test_fleet.py +++ b/src/tests/_internal/cli/utils/test_fleet.py @@ -3,6 +3,7 @@ from typing import List, Optional from uuid import uuid4 +import pytest from rich.table import Table from rich.text import Text @@ -24,6 +25,7 @@ Gpu, Instance, InstanceStatus, + InstanceTerminationReason, InstanceType, Resources, SSHKey, @@ -214,6 +216,22 @@ def create_ssh_fleet( class TestGetFleetsTable: + @pytest.mark.parametrize( + ("message", "expected"), + [("No quota", "no quota"), ("No balance", "no balance"), ("Other error", "no_offers")], + ) + def test_blocked_offers_with_legacy_reason(self, message, expected): + instance = create_test_instance(status=InstanceStatus.TERMINATED) + instance.termination_reason = InstanceTerminationReason.NO_OFFERS.value + instance.termination_reason_message = message + fleet = create_backend_fleet(instances=[instance]) + + cells = get_table_cells( + get_fleets_table([fleet], current_project="test-project", verbose=True) + ) + + assert cells[1]["ERROR"] == expected + def test_backend_fleet_without_verbose(self): instance = create_test_instance( instance_num=0, diff --git a/src/tests/_internal/cli/utils/test_run.py b/src/tests/_internal/cli/utils/test_run.py index dffde505e8..0d643dab81 100644 --- a/src/tests/_internal/cli/utils/test_run.py +++ b/src/tests/_internal/cli/utils/test_run.py @@ -11,6 +11,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload +from dstack._internal.cli.services.configurators.run import print_finished_message from dstack._internal.cli.utils.run import _format_pull_progress, get_runs_table from dstack._internal.core.models.backends.base import BackendType from dstack._internal.core.models.configurations import ( @@ -258,6 +259,22 @@ async def test_simple_run(self, session: AsyncSession): "no capacity", "gold1", ), + ( + JobStatus.FAILED, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA, + None, + "All matching offers are blocked by quota", + "no quota", + "indian_red1", + ), + ( + JobStatus.FAILED, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE, + None, + "All matching offers are blocked by insufficient balance", + "no balance", + "indian_red1", + ), ( JobStatus.FAILED, JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, @@ -579,3 +596,34 @@ async def test_service_with_multiple_replicas_and_jobs(self, session: AsyncSessi ) def test_format_pull_progress(progress: ImagePullProgress, expected: str) -> None: assert _format_pull_progress(progress) == expected + + +@pytest.mark.asyncio +@pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True) +@pytest.mark.usefixtures("image_config_mock") +class TestPrintFinishedMessage: + @pytest.mark.parametrize( + ("reason", "status_message"), + [ + (JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, "No capacity"), + (JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA, "No quota"), + (JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE, "No balance"), + ], + ) + async def test_provisioning_failure_has_no_runner_logs_hint( + self, + test_db, + session: AsyncSession, + capsys: pytest.CaptureFixture[str], + reason: JobTerminationReason, + status_message: str, + ): + run = await create_run_with_job( + session=session, job_status=JobStatus.FAILED, termination_reason=reason + ) + + print_finished_message(run) + + output = capsys.readouterr().out + assert status_message in output + assert "dstack logs" not in output diff --git a/src/tests/_internal/core/models/test_runs.py b/src/tests/_internal/core/models/test_runs.py index e911a07125..105d6d86b8 100644 --- a/src/tests/_internal/core/models/test_runs.py +++ b/src/tests/_internal/core/models/test_runs.py @@ -78,11 +78,26 @@ def test_job_termination_reason_to_retry_event_works_with_all_enum_variants(): assert retry_event is None or isinstance(retry_event, RetryEvent) +class TestJobTerminationReason: + @pytest.mark.parametrize( + "reason", + [ + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE, + ], + ) + def test_blocked_offers_preserve_no_capacity_retry_policy(self, reason): + assert reason.to_status() == JobStatus.FAILED + assert reason.to_retry_event() == RetryEvent.NO_CAPACITY + + # Will fail if JobTerminationReason value is added without updating JobSubmission._get_error def test_get_error_returns_expected_messages(): # already handled and shown in status_message no_error_reasons = [ JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE, JobTerminationReason.INTERRUPTED_BY_NO_CAPACITY, JobTerminationReason.WAITING_RUNNER_LIMIT_EXCEEDED, JobTerminationReason.TERMINATED_BY_USER, diff --git a/src/tests/_internal/server/background/pipeline_tasks/test_fleets.py b/src/tests/_internal/server/background/pipeline_tasks/test_fleets.py index 2268ad9400..47e2c92b76 100644 --- a/src/tests/_internal/server/background/pipeline_tasks/test_fleets.py +++ b/src/tests/_internal/server/background/pipeline_tasks/test_fleets.py @@ -693,8 +693,20 @@ async def test_prefers_surviving_instance_over_new_replacement_for_master_electi for instance in non_deleted_instances ) + @pytest.mark.parametrize( + "reason", + [ + InstanceTerminationReason.NO_OFFERS, + InstanceTerminationReason.NO_QUOTA, + InstanceTerminationReason.NO_BALANCE, + ], + ) async def test_min_zero_failed_master_terminates_unprovisioned_siblings( - self, test_db, session: AsyncSession, worker: FleetWorker + self, + test_db, + session: AsyncSession, + worker: FleetWorker, + reason: InstanceTerminationReason, ): project = await create_project(session) fleet = await create_fleet( @@ -716,7 +728,7 @@ async def test_min_zero_failed_master_terminates_unprovisioned_siblings( offer=None, instance_num=0, ) - failed_master.termination_reason = InstanceTerminationReason.NO_OFFERS + failed_master.termination_reason = reason sibling1 = await create_instance( session=session, project=project, diff --git a/src/tests/_internal/server/background/pipeline_tasks/test_instances/test_cloud_provisioning.py b/src/tests/_internal/server/background/pipeline_tasks/test_instances/test_cloud_provisioning.py index 45f29c8976..4d0ec29f4c 100644 --- a/src/tests/_internal/server/background/pipeline_tasks/test_instances/test_cloud_provisioning.py +++ b/src/tests/_internal/server/background/pipeline_tasks/test_instances/test_cloud_provisioning.py @@ -108,7 +108,10 @@ async def test_creates_instance( total_blocks=expected_blocks, ) backend_mock.compute.return_value = Mock(spec=ComputeMockSpec) - backend_mock.compute.return_value.get_offers.return_value = [offer] + backend_mock.compute.return_value.get_offers.return_value = [ + offer.model_copy(update={"availability": InstanceAvailability.NO_QUOTA}), + offer, + ] backend_mock.compute.return_value.create_instance.return_value = JobProvisioningData( backend=offer.backend, instance_type=offer.instance, @@ -130,6 +133,7 @@ async def test_creates_instance( assert instance.status == InstanceStatus.PROVISIONING assert instance.total_blocks == expected_blocks assert instance.busy_blocks == 0 + backend_mock.compute.return_value.create_instance.assert_called_once() @pytest.mark.parametrize("err", [RuntimeError("Unexpected"), ProvisioningError("Expected")]) async def test_tries_second_offer_if_first_fails( @@ -218,6 +222,57 @@ async def test_fails_if_no_offers( assert instance.status == InstanceStatus.TERMINATED assert instance.termination_reason == InstanceTerminationReason.NO_OFFERS + @pytest.mark.parametrize( + ("availabilities", "message"), + [ + ([InstanceAvailability.NO_QUOTA] * 2, "No quota"), + ( + [InstanceAvailability.NO_QUOTA, InstanceAvailability.NOT_AVAILABLE], + "No offers found", + ), + ([InstanceAvailability.NO_BALANCE] * 2, "No balance"), + ( + [InstanceAvailability.NO_QUOTA, InstanceAvailability.NO_BALANCE], + "No offers found", + ), + ( + [InstanceAvailability.NO_BALANCE, InstanceAvailability.NOT_AVAILABLE], + "No offers found", + ), + ], + ) + async def test_fails_without_attempting_unavailable_offers( + self, + test_db, + session: AsyncSession, + worker: InstanceWorker, + availabilities: list[InstanceAvailability], + message: str, + ): + project = await create_project(session=session) + instance = await create_instance( + session=session, project=project, status=InstanceStatus.PENDING + ) + backend = Mock(TYPE=BackendType.DSTACK) + compute = Mock(spec=ComputeMockSpec) + backend.compute.return_value = compute + compute.get_offers.return_value = [ + get_instance_offer_with_availability(availability=availability) + for availability in availabilities + ] + with patch( + "dstack._internal.server.services.backends.get_project_backends", + return_value=[backend], + ): + await process_instance(session, worker, instance) + + await session.refresh(instance) + assert instance.status == InstanceStatus.TERMINATED + assert instance.termination_reason == InstanceTerminationReason.NO_OFFERS + assert instance.termination_reason_message == message + compute.get_offers.assert_called_once() + compute.create_instance.assert_not_called() + async def test_waits_when_fleet_has_no_current_master( self, test_db, diff --git a/src/tests/_internal/server/background/pipeline_tasks/test_runs/test_active.py b/src/tests/_internal/server/background/pipeline_tasks/test_runs/test_active.py index b2c2971725..add2e30487 100644 --- a/src/tests/_internal/server/background/pipeline_tasks/test_runs/test_active.py +++ b/src/tests/_internal/server/background/pipeline_tasks/test_runs/test_active.py @@ -558,8 +558,16 @@ async def test_retries_no_capacity_replica_and_keeps_service_running( assert retried_job.status == JobStatus.SUBMITTED assert len(jobs) == 3 + @pytest.mark.parametrize( + "reason", + [ + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE, + ], + ) async def test_replica_retry_deletes_superseded_no_capacity_submissions( - self, test_db, session: AsyncSession, worker: RunWorker + self, test_db, session: AsyncSession, worker: RunWorker, reason: JobTerminationReason ) -> None: project = await create_project(session=session) user = await create_user(session=session) @@ -597,7 +605,7 @@ async def test_replica_retry_deletes_superseded_no_capacity_submissions( session=session, run=run, status=JobStatus.FAILED, - termination_reason=JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, + termination_reason=reason, submitted_at=run.submitted_at, last_processed_at=run.last_processed_at, replica_num=0, @@ -638,8 +646,16 @@ async def test_replica_retry_deletes_superseded_no_capacity_submissions( assert interrupted_job.id in [j.id for j in jobs] assert jobs[-1].status == JobStatus.SUBMITTED + @pytest.mark.parametrize( + "reason", + [ + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE, + ], + ) async def test_retries_scheduled_run_no_capacity_from_trigger_time( - self, test_db, session: AsyncSession, worker: RunWorker + self, test_db, session: AsyncSession, worker: RunWorker, reason: JobTerminationReason ) -> None: project = await create_project(session=session) user = await create_user(session=session) @@ -671,7 +687,7 @@ async def test_retries_scheduled_run_no_capacity_from_trigger_time( session=session, run=run, status=JobStatus.FAILED, - termination_reason=JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, + termination_reason=reason, ) lock_run(run) await session.commit() diff --git a/src/tests/_internal/server/background/pipeline_tasks/test_runs/test_pending.py b/src/tests/_internal/server/background/pipeline_tasks/test_runs/test_pending.py index 6cb006821d..163f62ee2b 100644 --- a/src/tests/_internal/server/background/pipeline_tasks/test_runs/test_pending.py +++ b/src/tests/_internal/server/background/pipeline_tasks/test_runs/test_pending.py @@ -154,8 +154,16 @@ async def test_resubmits_retrying_run_after_delay( assert new_job.replica_num == old_job.replica_num assert new_job.submission_num == old_job.submission_num + 1 + @pytest.mark.parametrize( + "reason", + [ + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE, + ], + ) async def test_resubmission_deletes_superseded_no_capacity_submissions( - self, test_db, session: AsyncSession, worker: RunWorker + self, test_db, session: AsyncSession, worker: RunWorker, reason: JobTerminationReason ) -> None: """ On resubmission, older submissions that failed to start due to no capacity @@ -180,7 +188,7 @@ async def test_resubmission_deletes_superseded_no_capacity_submissions( status=JobStatus.FAILED, last_processed_at=old_time, submission_num=0, - termination_reason=JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, + termination_reason=reason, ) provisioned_job = await create_job( session=session, @@ -188,7 +196,7 @@ async def test_resubmission_deletes_superseded_no_capacity_submissions( status=JobStatus.FAILED, last_processed_at=old_time, submission_num=1, - termination_reason=JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, + termination_reason=reason, job_provisioning_data=get_job_provisioning_data(), ) other_reason_job = await create_job( @@ -205,7 +213,7 @@ async def test_resubmission_deletes_superseded_no_capacity_submissions( status=JobStatus.FAILED, last_processed_at=old_time, submission_num=3, - termination_reason=JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, + termination_reason=reason, ) lock_run(run) await session.commit() diff --git a/src/tests/_internal/server/background/pipeline_tasks/test_submitted_jobs.py b/src/tests/_internal/server/background/pipeline_tasks/test_submitted_jobs.py index 536df52cc1..befd73ef86 100644 --- a/src/tests/_internal/server/background/pipeline_tasks/test_submitted_jobs.py +++ b/src/tests/_internal/server/background/pipeline_tasks/test_submitted_jobs.py @@ -6,7 +6,7 @@ import gpuhunt import pytest -from sqlalchemy import select +from sqlalchemy import event, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import joinedload @@ -25,7 +25,7 @@ ) from dstack._internal.core.models.envs import Env from dstack._internal.core.models.fleets import FleetNodesSpec, InstanceGroupPlacement -from dstack._internal.core.models.instances import InstanceStatus +from dstack._internal.core.models.instances import InstanceAvailability, InstanceStatus from dstack._internal.core.models.placement import PlacementGroup from dstack._internal.core.models.profiles import ( CreationPolicy, @@ -431,7 +431,10 @@ async def test_provisions_new_capacity_for_assigned_job_with_placeholder( backend_mock = Mock() m.return_value = [backend_mock] backend_mock.TYPE = BackendType.AWS - backend_mock.compute.return_value.get_offers.return_value = [offer] + backend_mock.compute.return_value.get_offers.return_value = [ + offer.model_copy(update={"availability": InstanceAvailability.NO_QUOTA}), + offer, + ] backend_mock.compute.return_value.run_job.return_value = get_job_provisioning_data( dockerized=True, backend=BackendType.AWS, @@ -458,6 +461,7 @@ async def test_provisions_new_capacity_for_assigned_job_with_placeholder( assert job.instance.fleet_id == fleet.id assert job.instance.offer is not None assert job.instance.provisioning_job_id == job.id # never cleared + backend_mock.compute.return_value.run_job.assert_called_once() res = await session.execute( select(InstanceModel).where( InstanceModel.fleet_id == fleet.id, @@ -2037,8 +2041,15 @@ async def test_leaves_placeholder_for_terminating_pipeline_on_failed_new_capacit assert not placeholder.deleted assert placeholder.status == InstanceStatus.PENDING + @pytest.mark.parametrize( + "blocked_availability", [InstanceAvailability.NO_QUOTA, InstanceAvailability.NO_BALANCE] + ) async def test_reports_tried_offers_when_new_capacity_provisioning_fails( - self, test_db, session: AsyncSession, worker: JobSubmittedWorker + self, + test_db, + session: AsyncSession, + worker: JobSubmittedWorker, + blocked_availability: InstanceAvailability, ): project = await create_project(session=session) user = await create_user(session=session) @@ -2056,7 +2067,10 @@ async def test_reports_tried_offers_when_new_capacity_provisioning_fails( backend_mock.TYPE = BackendType.AWS backend_mock.compute.return_value = compute_mock m.return_value = [backend_mock] - compute_mock.get_offers.return_value = [offer] + compute_mock.get_offers.return_value = [ + offer.model_copy(update={"availability": blocked_availability}), + offer, + ] compute_mock.run_job.side_effect = BackendError("InsufficientInstanceCapacity") # The first pass assigns the job to the fleet, the second one provisions. @@ -2069,6 +2083,151 @@ async def test_reports_tried_offers_when_new_capacity_provisioning_fails( assert f"Failed to provision in fleet '{fleet.name}'" in job.termination_reason_message assert "tried 1 of 1 offers, all failed" in job.termination_reason_message assert "us-east-1: InsufficientInstanceCapacity" in job.termination_reason_message + compute_mock.run_job.assert_called_once() + + @pytest.mark.parametrize( + ("availabilities", "message"), + [ + ([], None), + ( + [InstanceAvailability.NO_QUOTA] * 2, + "No quota", + ), + ( + [InstanceAvailability.NO_QUOTA, InstanceAvailability.NOT_AVAILABLE], + None, + ), + ( + [InstanceAvailability.NO_BALANCE], + "No balance", + ), + ( + [InstanceAvailability.NO_QUOTA, InstanceAvailability.NO_BALANCE], + None, + ), + ( + [InstanceAvailability.NO_BALANCE, InstanceAvailability.NOT_AVAILABLE], + None, + ), + ], + ) + async def test_fails_without_attempting_unavailable_offers( + self, + test_db, + session: AsyncSession, + worker: JobSubmittedWorker, + availabilities: list[InstanceAvailability], + message: str | None, + ): + project = await create_project(session=session) + user = await create_user(session=session) + repo = await create_repo(session=session, project_id=project.id) + fleet = await create_fleet(session=session, project=project) + run = await create_run(session=session, project=project, repo=repo, user=user, fleet=fleet) + job = await create_job(session=session, run=run) + backend = Mock(TYPE=BackendType.DSTACK) + compute = Mock(spec=ComputeMockSpec) + backend.compute.return_value = compute + compute.get_offers.return_value = [ + get_instance_offer_with_availability(availability=availability) + for availability in availabilities + ] + with patch( + "dstack._internal.server.services.backends.get_project_backends", + return_value=[backend], + ): + await _process_job(session=session, worker=worker, job_model=job) + await _process_job(session=session, worker=worker, job_model=job) + + job = await _get_job(session, job.id) + assert job.status == JobStatus.TERMINATING + assert job.termination_reason == JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY + if message is not None: + assert job.termination_reason_message == message + else: + assert job.termination_reason_message is not None + assert "No offers matching the run requirements" in job.termination_reason_message + assert job.lock_token is None + assert job.instance is not None + assert not job.instance.deleted + assert job.instance.status == InstanceStatus.PENDING + compute.get_offers.assert_called_once() + compute.run_job.assert_not_called() + compute.run_jobs.assert_not_called() + compute.create_instance.assert_not_called() + + async def test_blocked_offers_keep_query_count_and_offer_iteration_outside_sessions( + self, test_db, session: AsyncSession, worker: JobSubmittedWorker + ): + user = await create_user(session=session) + query_counts = [] + for index, availability in enumerate( + [None, InstanceAvailability.NO_QUOTA, InstanceAvailability.NO_BALANCE] + ): + project = await create_project(session=session, name=f"project_{index}", owner=user) + repo = await create_repo(session=session, project_id=project.id) + fleet = await create_fleet(session=session, project=project) + run = await create_run( + session=session, project=project, repo=repo, user=user, fleet=fleet + ) + job = await create_job(session=session, run=run) + await session.commit() + active_connections = 0 + queries = 0 + + def checkout(*args): + nonlocal active_connections + active_connections += 1 + + def checkin(*args): + nonlocal active_connections + active_connections -= 1 + + def before_cursor_execute(*args): + nonlocal queries + queries += 1 + + def iterate_offers(): + if availability is None: + return + offer = get_instance_offer_with_availability(availability=availability) + for _ in range(1000): + # The generator resumes after each offer's rejection is processed. + assert active_connections == 0 + previous_queries = queries + yield offer + assert queries == previous_queries + assert active_connections == 0 + + backend = Mock(TYPE=BackendType.DSTACK) + backend.compute.return_value.get_offers.side_effect = lambda *args: iterate_offers() + engine = test_db.engine.sync_engine + listeners = [ + ("checkout", checkout), + ("checkin", checkin), + ("before_cursor_execute", before_cursor_execute), + ] + for name, listener in listeners: + event.listen(engine, name, listener) + try: + with patch( + "dstack._internal.server.services.backends.get_project_backends", + return_value=[backend], + ): + await _process_job(session=session, worker=worker, job_model=job) + await _process_job(session=session, worker=worker, job_model=job) + query_counts.append(queries) + assert active_connections == 0 + finally: + for name, listener in listeners: + event.remove(engine, name, listener) + job = await _get_job(session, job.id) + assert ( + job.termination_reason == JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY + ) + + assert query_counts[0] > 0 + assert len(set(query_counts)) == 1 async def test_provisions_compute_group( self, test_db, session: AsyncSession, worker: JobSubmittedWorker diff --git a/src/tests/_internal/server/services/jobs/test_jobs.py b/src/tests/_internal/server/services/jobs/test_jobs.py index 96b127e823..9efddbc73d 100644 --- a/src/tests/_internal/server/services/jobs/test_jobs.py +++ b/src/tests/_internal/server/services/jobs/test_jobs.py @@ -10,11 +10,13 @@ from dstack._internal.core.models.profiles import Profile from dstack._internal.core.models.repos.local import LocalRunRepoData from dstack._internal.core.models.resources import ResourcesSpec -from dstack._internal.core.models.runs import JobSpec, RunSpec +from dstack._internal.core.models.runs import JobSpec, JobStatus, JobTerminationReason, RunSpec from dstack._internal.core.models.volumes import DaytonaVolumeConfiguration +from dstack._internal.server.models import JobModel from dstack._internal.server.services.docker import ImageConfig from dstack._internal.server.services.jobs import ( _get_job_mount_point_attached_volume, + _get_job_status_message, get_job_specs_from_run_spec, job_spec_updatable_in_place, ) @@ -26,6 +28,34 @@ ) +class TestGetJobStatusMessage: + @pytest.mark.parametrize( + ("reason", "message"), + [ + (JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA, "no quota"), + (JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE, "no balance"), + ], + ) + def test_reports_blocked_offers(self, reason, message): + job = JobModel( + status=JobStatus.FAILED, + termination_reason=reason, + ) + assert _get_job_status_message(job) == message + + @pytest.mark.parametrize( + ("message", "expected"), + [("No quota", "no quota"), ("No balance", "no balance"), ("Other error", "no capacity")], + ) + def test_reports_blocked_offers_with_legacy_reason(self, message, expected): + job = JobModel( + status=JobStatus.FAILED, + termination_reason=JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, + termination_reason_message=message, + ) + assert _get_job_status_message(job) == expected + + @pytest.mark.parametrize( "configuration, expected_calls", [ diff --git a/src/tests/_internal/server/services/runs/test_plan.py b/src/tests/_internal/server/services/runs/test_plan.py index e55ff6de67..e943eb40df 100644 --- a/src/tests/_internal/server/services/runs/test_plan.py +++ b/src/tests/_internal/server/services/runs/test_plan.py @@ -24,6 +24,7 @@ ) from dstack._internal.core.models.resources import CPUSpec, GPUSpec, Memory, Range, ResourcesSpec from dstack._internal.server.services.jobs import get_jobs_from_run_spec +from dstack._internal.server.services.offers import Offers from dstack._internal.server.services.projects import get_project_model_by_name from dstack._internal.server.services.runs import get_plan from dstack._internal.server.services.runs.plan import ( @@ -837,7 +838,7 @@ async def test_keeps_unconstrained_offers_for_non_empty_cluster_fleet_without_el ) offer = get_instance_offer_with_availability() backend = AsyncMock() - get_offers_by_requirements_mock.return_value = [(backend, offer)] + get_offers_by_requirements_mock.return_value = Offers(offers=[(backend, offer)]) offers = await _get_backend_offers_in_fleet( project=project, diff --git a/src/tests/_internal/server/services/runs/test_runs.py b/src/tests/_internal/server/services/runs/test_runs.py index cc283a6812..6c624f1353 100644 --- a/src/tests/_internal/server/services/runs/test_runs.py +++ b/src/tests/_internal/server/services/runs/test_runs.py @@ -27,11 +27,20 @@ class TestListUserRuns: @pytest.mark.asyncio @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True) @pytest.mark.parametrize("include_jobs", [True, False]) + @pytest.mark.parametrize( + "termination_reason", + [ + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA, + JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE, + ], + ) async def test_limited_list_materializes_only_latest_and_status_jobs( self, test_db, session: AsyncSession, include_jobs: bool, + termination_reason: JobTerminationReason, monkeypatch: pytest.MonkeyPatch, ) -> None: user = await create_user(session=session, global_role=GlobalRole.USER) @@ -60,7 +69,7 @@ async def test_limited_list_materializes_only_latest_and_status_jobs( run=run, submission_num=0, status=JobStatus.TERMINATED, - termination_reason=JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, + termination_reason=termination_reason, ) for submission_num in range(1, 12): await create_job( diff --git a/src/tests/_internal/server/services/test_offers.py b/src/tests/_internal/server/services/test_offers.py index 7ff0b4fe4a..fc6bdeac2f 100644 --- a/src/tests/_internal/server/services/test_offers.py +++ b/src/tests/_internal/server/services/test_offers.py @@ -3,20 +3,296 @@ import pytest from dstack._internal.core.models.backends.base import BackendType +from dstack._internal.core.models.instances import ( + InstanceAvailability, + InstanceOfferWithAvailability, +) +from dstack._internal.core.models.placement import PlacementGroup from dstack._internal.core.models.profiles import Profile from dstack._internal.core.models.resources import ResourcesSpec from dstack._internal.core.models.runs import Requirements from dstack._internal.core.models.volumes import DaytonaVolumeConfiguration from dstack._internal.server.services.offers import get_offers_by_requirements from dstack._internal.server.testing.common import ( + ComputeMockSpec, get_instance_offer_with_availability, get_kubernetes_volume_configuration, + get_placement_group_configuration, get_volume, get_volume_configuration, ) class TestGetOffersByRequirements: + @pytest.mark.asyncio + @pytest.mark.parametrize("exclude_not_available", [False, True]) + async def test_stops_at_max_offers(self, exclude_not_available): + offer = get_instance_offer_with_availability() + backend = Mock(TYPE=BackendType.AWS) + + def iterate_offers(): + yield offer + raise AssertionError("Planning consumed offers beyond max_offers") + + backend.compute.return_value.get_offers.return_value = iterate_offers() + with patch( + "dstack._internal.server.services.backends.get_project_backends", + return_value=[backend], + ): + result = await get_offers_by_requirements( + project=Mock(), + profile=Profile(), + requirements=Requirements(resources=ResourcesSpec()), + max_offers=1, + exclude_not_available=exclude_not_available, + ) + assert result.offers == [(backend, offer)] + + @pytest.mark.asyncio + @pytest.mark.parametrize( + ("availabilities", "no_quota", "no_balance"), + [ + ([], False, False), + ([InstanceAvailability.NO_QUOTA], True, False), + ([InstanceAvailability.NO_BALANCE], False, True), + ([InstanceAvailability.NO_QUOTA, InstanceAvailability.NOT_AVAILABLE], False, False), + ([InstanceAvailability.NO_BALANCE, InstanceAvailability.NOT_AVAILABLE], False, False), + ([InstanceAvailability.NO_QUOTA, InstanceAvailability.NO_BALANCE], False, False), + ([InstanceAvailability.AVAILABLE, InstanceAvailability.NO_QUOTA], False, False), + ([InstanceAvailability.AVAILABLE, InstanceAvailability.NO_BALANCE], False, False), + ], + ) + async def test_collects_unavailable_offers_without_returning_them( + self, availabilities, no_quota, no_balance + ): + backend = Mock(TYPE=BackendType.DSTACK) + offers = [get_instance_offer_with_availability(availability=a) for a in availabilities] + backend.compute.return_value.get_offers.return_value = iter(offers) + with patch( + "dstack._internal.server.services.backends.get_project_backends", + return_value=[backend], + ): + result = await get_offers_by_requirements( + project=Mock(), + profile=Profile(), + requirements=Requirements(resources=ResourcesSpec()), + exclude_not_available=True, + ) + assert result.offers == [ + (backend, offer) for offer in offers if offer.availability.is_available() + ] + assert (result.unavailable_reason == InstanceAvailability.NO_QUOTA) == no_quota + assert (result.unavailable_reason == InstanceAvailability.NO_BALANCE) == no_balance + backend.compute.return_value.get_offers.assert_called_once() + + @pytest.mark.asyncio + @pytest.mark.parametrize( + ("profile", "ignored_offer", "blocks", "supported_backends"), + [ + ( + Profile(backends=[BackendType.AWS]), + get_instance_offer_with_availability(backend=BackendType.GCP), + 1, + None, + ), + ( + Profile(regions=["eu-west"]), + get_instance_offer_with_availability(region="us-east"), + 1, + None, + ), + ( + Profile(instance_types=["instance"]), + get_instance_offer_with_availability(instance_type="other"), + 1, + None, + ), + ( + Profile(availability_zones=["AZ1"]), + get_instance_offer_with_availability(availability_zones=["az2"]), + 1, + None, + ), + ( + Profile(availability_zones=["az1"]), + get_instance_offer_with_availability(availability_zones=None), + 1, + None, + ), + (Profile(), get_instance_offer_with_availability(cpu_count=3), 2, None), + ( + Profile(), + get_instance_offer_with_availability(backend=BackendType.KUBERNETES), + 1, + [BackendType.AWS], + ), + ], + ) + async def test_ignores_unavailable_offers_that_do_not_match( + self, + profile: Profile, + ignored_offer: InstanceOfferWithAvailability, + blocks, + supported_backends, + ): + quota_offer = get_instance_offer_with_availability( + availability=InstanceAvailability.NO_QUOTA, availability_zones=["az1"] + ) + ignored_offer = ignored_offer.model_copy( + update={"availability": InstanceAvailability.NOT_AVAILABLE} + ) + backend = Mock(TYPE=BackendType.DSTACK) + backend.compute.return_value.get_offers.return_value = [quota_offer, ignored_offer] + with patch( + "dstack._internal.server.services.backends.get_project_backends", + return_value=[backend], + ): + result = await get_offers_by_requirements( + project=Mock(), + profile=profile, + requirements=Requirements(resources=ResourcesSpec()), + exclude_not_available=True, + blocks=blocks, + supported_backend_types=supported_backends, + ) + assert result.offers == [] + assert result.unavailable_reason == InstanceAvailability.NO_QUOTA + + @pytest.mark.asyncio + async def test_does_not_copy_unavailable_offers_for_az_or_blocks(self): + offer = get_instance_offer_with_availability( + availability=InstanceAvailability.NO_QUOTA, + availability_zones=["AZ1", "az2"], + cpu_count=128, + gpu_count=8, + ) + backend = Mock(TYPE=BackendType.DSTACK) + backend.compute.return_value.get_offers.return_value = (offer for _ in range(1000)) + with ( + patch( + "dstack._internal.server.services.backends.get_project_backends", + return_value=[backend], + ), + patch.object(InstanceOfferWithAvailability, "model_copy", side_effect=AssertionError), + ): + result = await get_offers_by_requirements( + project=Mock(), + profile=Profile(availability_zones=["az1"]), + requirements=Requirements(resources=ResourcesSpec()), + blocks="auto", + exclude_not_available=True, + ) + assert result.offers == [] + assert result.unavailable_reason == InstanceAvailability.NO_QUOTA + assert offer.availability_zones == ["AZ1", "az2"] + assert offer.total_blocks == 1 + + @pytest.mark.asyncio + async def test_only_collects_unavailable_offers_in_the_volume_location(self): + backend = Mock(TYPE=BackendType.DSTACK) + backend.compute.return_value.get_offers.return_value = [ + get_instance_offer_with_availability(availability=InstanceAvailability.NO_QUOTA), + get_instance_offer_with_availability( + availability=InstanceAvailability.NOT_AVAILABLE, region="other-region" + ), + get_instance_offer_with_availability( + availability=InstanceAvailability.NOT_AVAILABLE, backend=BackendType.GCP + ), + ] + with patch( + "dstack._internal.server.services.backends.get_project_backends", + return_value=[backend], + ): + result = await get_offers_by_requirements( + project=Mock(), + profile=Profile(), + requirements=Requirements(resources=ResourcesSpec()), + exclude_not_available=True, + volumes=[[get_volume(configuration=get_volume_configuration(region="eu-west"))]], + ) + assert result.offers == [] + assert result.unavailable_reason == InstanceAvailability.NO_QUOTA + + @pytest.mark.asyncio + async def test_only_collects_unavailable_offers_suitable_for_the_placement_group(self): + backend = Mock(TYPE=BackendType.DSTACK) + compute = Mock(spec=ComputeMockSpec) + backend.compute.return_value = compute + compute.is_suitable_placement_group.side_effect = ( + lambda pg, offer: offer.region == pg.configuration.region + ) + compute.get_offers.return_value = [ + get_instance_offer_with_availability(availability=InstanceAvailability.NO_QUOTA), + get_instance_offer_with_availability( + availability=InstanceAvailability.NOT_AVAILABLE, region="other-region" + ), + ] + with patch( + "dstack._internal.server.services.backends.get_project_backends", + return_value=[backend], + ): + result = await get_offers_by_requirements( + project=Mock(), + profile=Profile(), + requirements=Requirements(resources=ResourcesSpec()), + exclude_not_available=True, + placement_group=PlacementGroup( + name="placement-group", + project_name="test_project", + configuration=get_placement_group_configuration(region="eu-west"), + ), + ) + assert result.offers == [] + assert result.unavailable_reason == InstanceAvailability.NO_QUOTA + + @pytest.mark.asyncio + @pytest.mark.parametrize("feature", ["multinode", "privileged", "instance_mounts"]) + async def test_only_collects_unavailable_offers_with_required_features(self, feature): + backend = Mock(TYPE=BackendType.DSTACK) + backend.compute.return_value.get_offers.return_value = [ + get_instance_offer_with_availability(availability=InstanceAvailability.NO_QUOTA), + get_instance_offer_with_availability( + backend=BackendType.DAYTONA, availability=InstanceAvailability.NOT_AVAILABLE + ), + ] + with patch( + "dstack._internal.server.services.backends.get_project_backends", + return_value=[backend], + ): + result = await get_offers_by_requirements( + project=Mock(), + profile=Profile(), + requirements=Requirements(resources=ResourcesSpec()), + exclude_not_available=True, + **{feature: True}, + ) + assert result.offers == [] + assert result.unavailable_reason == InstanceAvailability.NO_QUOTA + + @pytest.mark.asyncio + async def test_preserves_available_offer_order_when_a_backend_is_not_sorted_by_price(self): + vastai = Mock(TYPE=BackendType.VASTAI) + first = get_instance_offer_with_availability(backend=BackendType.VASTAI, price=1) + blocked = get_instance_offer_with_availability( + backend=BackendType.VASTAI, price=100, availability=InstanceAvailability.NO_QUOTA + ) + second = get_instance_offer_with_availability(backend=BackendType.VASTAI, price=2) + vastai.compute.return_value.get_offers.return_value = [first, blocked, second] + aws = Mock(TYPE=BackendType.AWS) + third = get_instance_offer_with_availability(price=3) + aws.compute.return_value.get_offers.return_value = [third] + with patch( + "dstack._internal.server.services.backends.get_project_backends", + return_value=[vastai, aws], + ): + result = await get_offers_by_requirements( + project=Mock(), + profile=Profile(), + requirements=Requirements(resources=ResourcesSpec()), + exclude_not_available=True, + ) + assert result.offers == [(vastai, first), (vastai, second), (aws, third)] + @pytest.mark.asyncio async def test_returns_all_offers(self): profile = Profile(name="test") @@ -37,7 +313,10 @@ async def test_returns_all_offers(self): requirements=requirements, ) m.assert_awaited_once() - assert res == [(aws_backend_mock, aws_offer), (runpod_backend_mock, runpod_offer)] + assert res.offers == [ + (aws_backend_mock, aws_offer), + (runpod_backend_mock, runpod_offer), + ] @pytest.mark.asyncio async def test_returns_multinode_offers(self): @@ -60,7 +339,7 @@ async def test_returns_multinode_offers(self): multinode=True, ) m.assert_awaited_once() - assert res == [(aws_backend_mock, aws_offer)] + assert res.offers == [(aws_backend_mock, aws_offer)] @pytest.mark.asyncio async def test_returns_volume_offers(self): @@ -99,7 +378,7 @@ async def test_returns_volume_offers(self): ], ) m.assert_awaited_once() - assert res == [(runpod_backend_mock, runpod_offer2)] + assert res.offers == [(runpod_backend_mock, runpod_offer2)] @pytest.mark.asyncio async def test_global_volume_matches_cpu_and_gpu_regions(self): @@ -124,7 +403,7 @@ async def test_global_volume_matches_cpu_and_gpu_regions(self): volumes=[[get_volume(configuration=DaytonaVolumeConfiguration())]], ) - assert offers == [(daytona_backend, cpu_offer), (daytona_backend, gpu_offer)] + assert offers.offers == [(daytona_backend, cpu_offer), (daytona_backend, gpu_offer)] @pytest.mark.asyncio async def test_returns_volume_offers_without_region(self): @@ -156,7 +435,7 @@ async def test_returns_volume_offers_without_region(self): volumes=[[get_volume(configuration=get_kubernetes_volume_configuration())]], ) m.assert_awaited_once() - assert res == [(kubernetes_backend_mock, kubernetes_offer)] + assert res.offers == [(kubernetes_backend_mock, kubernetes_offer)] @pytest.mark.asyncio async def test_returns_az_offers(self): @@ -192,7 +471,10 @@ async def test_returns_az_offers(self): requirements=requirements, ) m.assert_awaited_once() - assert res == [(aws_backend_mock, aws_offer1), (aws_backend_mock, expected_aws_offer3)] + assert res.offers == [ + (aws_backend_mock, aws_offer1), + (aws_backend_mock, expected_aws_offer3), + ] @pytest.mark.asyncio async def test_returns_az_offers_ignoring_case(self): @@ -214,7 +496,7 @@ async def test_returns_az_offers_ignoring_case(self): profile=profile, requirements=requirements, ) - assert res == [(aws_backend_mock, expected_aws_offer)] + assert res.offers == [(aws_backend_mock, expected_aws_offer)] @pytest.mark.asyncio async def test_returns_no_offers_for_multinode_instance_mounts_and_non_multinode_backend(self): @@ -239,4 +521,4 @@ async def test_returns_no_offers_for_multinode_instance_mounts_and_non_multinode instance_mounts=True, ) m.assert_awaited_once() - assert res == [] + assert res.offers == [] From 5e547cb76acf9854a08d0c6b169dd1b064d3c12c Mon Sep 17 00:00:00 2001 From: Andrey Cheptsov Date: Thu, 22 Oct 2026 15:58:22 -0700 Subject: [PATCH 2/2] Simplify quota and balance failure diagnosis --- src/dstack/_internal/cli/utils/fleet.py | 4 +- src/dstack/_internal/core/consts.py | 4 + src/dstack/_internal/core/models/instances.py | 2 +- src/dstack/_internal/core/models/runs.py | 2 +- .../background/pipeline_tasks/fleets.py | 4 +- .../instances/cloud_provisioning.py | 86 +++-- .../pipeline_tasks/jobs_submitted.py | 101 ++++-- .../server/services/backends/__init__.py | 20 +- .../_internal/server/services/fleets.py | 17 +- .../server/services/jobs/__init__.py | 8 +- .../_internal/server/services/offers.py | 224 +++++-------- .../_internal/server/services/runs/plan.py | 6 +- src/tests/_internal/cli/utils/test_run.py | 32 -- src/tests/_internal/core/models/test_runs.py | 13 - .../test_instances/test_cloud_provisioning.py | 20 +- .../pipeline_tasks/test_runs/test_active.py | 24 +- .../pipeline_tasks/test_submitted_jobs.py | 133 ++------ .../server/services/jobs/test_jobs.py | 14 - .../server/services/runs/test_plan.py | 3 +- .../_internal/server/services/test_offers.py | 298 +----------------- 20 files changed, 316 insertions(+), 699 deletions(-) diff --git a/src/dstack/_internal/cli/utils/fleet.py b/src/dstack/_internal/cli/utils/fleet.py index 6963e493be..666ebf2f19 100644 --- a/src/dstack/_internal/cli/utils/fleet.py +++ b/src/dstack/_internal/cli/utils/fleet.py @@ -8,6 +8,7 @@ format_backend, format_entity_reference, ) +from dstack._internal.core.consts import NO_BALANCE_MESSAGE, NO_QUOTA_MESSAGE from dstack._internal.core.models.backends.base import BackendType from dstack._internal.core.models.fleets import Fleet, FleetNodesSpec, FleetStatus from dstack._internal.core.models.health import HealthStatus @@ -172,7 +173,8 @@ def get_fleets_table( instance_row["ERROR"] = instance.termination_reason if ( instance.termination_reason == InstanceTerminationReason.NO_OFFERS - and instance.termination_reason_message in ("No quota", "No balance") + and instance.termination_reason_message + in (NO_QUOTA_MESSAGE, NO_BALANCE_MESSAGE) ): instance_row["ERROR"] = instance.termination_reason_message.lower() diff --git a/src/dstack/_internal/core/consts.py b/src/dstack/_internal/core/consts.py index 9b60381bdc..65561d78bf 100644 --- a/src/dstack/_internal/core/consts.py +++ b/src/dstack/_internal/core/consts.py @@ -12,5 +12,9 @@ DSTACK_PROJECT_ENV = "DSTACK_PROJECT" DSTACK_SERVER_URL_ENV = "DSTACK_SERVER_URL" DSTACK_TOKEN_ENV = "DSTACK_TOKEN" +# Keep these messages stable while existing termination reasons carry quota/balance failures +# during rolling upgrades. Status formatting recognizes their exact values. +NO_QUOTA_MESSAGE = "No quota" +NO_BALANCE_MESSAGE = "No balance" # legacy AWS, Azure, GCP, and OCI image for older GPUs DSTACK_OS_IMAGE_WITH_PROPRIETARY_NVIDIA_KERNEL_MODULES = "0.10" diff --git a/src/dstack/_internal/core/models/instances.py b/src/dstack/_internal/core/models/instances.py index e0fdfa227c..bd14defdd7 100644 --- a/src/dstack/_internal/core/models/instances.py +++ b/src/dstack/_internal/core/models/instances.py @@ -246,7 +246,7 @@ class InstanceTerminationReason(str, Enum): JOB_FINISHED = "job_finished" UNREACHABLE = "unreachable" NO_OFFERS = "no_offers" - # Add read support before emission so server replicas can be upgraded gradually. + # Recognize new reasons before storing them to support rolling upgrades. NO_QUOTA = "no_quota" MASTER_FAILED = "master_failed" MAX_INSTANCES_LIMIT = "max_instances_limit" diff --git a/src/dstack/_internal/core/models/runs.py b/src/dstack/_internal/core/models/runs.py index 032e6bea65..f7525330a2 100644 --- a/src/dstack/_internal/core/models/runs.py +++ b/src/dstack/_internal/core/models/runs.py @@ -138,7 +138,7 @@ def to_error(self) -> Optional[str]: class JobTerminationReason(str, Enum): # Set by the server FAILED_TO_START_DUE_TO_NO_CAPACITY = "failed_to_start_due_to_no_capacity" - # Add read support before emission so server replicas can be upgraded gradually. + # Recognize new reasons before storing them to support rolling upgrades. FAILED_TO_START_DUE_TO_NO_QUOTA = "failed_to_start_due_to_no_quota" FAILED_TO_START_DUE_TO_NO_BALANCE = "failed_to_start_due_to_no_balance" INTERRUPTED_BY_NO_CAPACITY = "interrupted_by_no_capacity" diff --git a/src/dstack/_internal/server/background/pipeline_tasks/fleets.py b/src/dstack/_internal/server/background/pipeline_tasks/fleets.py index ff285ae5eb..57d2896789 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/fleets.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/fleets.py @@ -819,8 +819,8 @@ def _set_fail_instances_on_master_bootstrap_failure( instance_id_to_update_map: dict[uuid.UUID, _InstanceUpdateMap], ) -> None: """ - Terminates unprovisioned siblings when the cluster master has no usable offers. - For nodes.min == 0, this prevents a master re-election loop. + Terminates instances with MASTER_FAILED if the master dies with NO_OFFERS in a cluster with node.min == 0. + This is needed to avoid master re-election loop and fail fast. """ fleet_spec = get_fleet_spec(fleet_model) if ( diff --git a/src/dstack/_internal/server/background/pipeline_tasks/instances/cloud_provisioning.py b/src/dstack/_internal/server/background/pipeline_tasks/instances/cloud_provisioning.py index de8e605c37..cead273c5e 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/instances/cloud_provisioning.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/instances/cloud_provisioning.py @@ -23,13 +23,13 @@ SkipOffer, ) from dstack._internal.core.models.instances import ( - InstanceAvailability, InstanceOfferWithAvailability, InstanceStatus, InstanceTerminationReason, ) from dstack._internal.core.models.placement import PlacementGroupConfiguration, PlacementStrategy -from dstack._internal.core.models.runs import JobProvisioningData +from dstack._internal.core.models.profiles import Profile +from dstack._internal.core.models.runs import JobProvisioningData, Requirements from dstack._internal.server import settings as server_settings from dstack._internal.server.background.pipeline_tasks.base import NOW_PLACEHOLDER from dstack._internal.server.background.pipeline_tasks.instances.common import ( @@ -46,7 +46,10 @@ get_instance_requirements, ) from dstack._internal.server.services.logging import fmt -from dstack._internal.server.services.offers import get_instance_offer_with_restricted_az +from dstack._internal.server.services.offers import ( + get_instance_offer_with_restricted_az, + get_offers_termination_reason_message, +) from dstack._internal.server.services.placement import ( can_use_placement_groups, get_fleet_placement_group_models, @@ -117,24 +120,7 @@ async def create_cloud_instance(instance_model: InstanceModel) -> ProcessResult: include_only_create_instance_supported_backends=True, ) - available_offers = offers.offers - for availability, message in ( - (InstanceAvailability.NO_QUOTA, "No quota"), - (InstanceAvailability.NO_BALANCE, "No balance"), - ): - if offers.unavailable_reason == availability: - set_status_update( - update_map=result.instance_update_map, - instance_model=instance_model, - new_status=InstanceStatus.TERMINATED, - termination_reason=InstanceTerminationReason.NO_OFFERS, - # TODO: Store NO_QUOTA/NO_BALANCE once the minimum supported rolling-upgrade - # source version handles them on every replica. - termination_reason_message=message, - ) - return result - - offers_iter = iter(available_offers) + offers_iter = iter(offers) offers_tried = 0 # Limit number of offers tried to prevent long-running processing in case all offers fail. while offers_tried < server_settings.MAX_OFFERS_TRIED: @@ -254,16 +240,72 @@ async def create_cloud_instance(instance_model: InstanceModel) -> ProcessResult: result.schedule_pg_deletion_except_id = placement_group_model.id return result + termination_reason_message = "All offers failed" if offers else "No offers found" + if not offers: + termination_reason_message = ( + await _get_no_offers_termination_reason_message( + instance_model=instance_model, + profile=profile, + requirements=requirements, + placement_group_model=placement_group_model, + master_job_provisioning_data=master_job_provisioning_data, + ) + or termination_reason_message + ) + set_status_update( update_map=result.instance_update_map, instance_model=instance_model, new_status=InstanceStatus.TERMINATED, termination_reason=InstanceTerminationReason.NO_OFFERS, - termination_reason_message="All offers failed" if available_offers else "No offers found", + termination_reason_message=termination_reason_message, ) return result +async def _get_no_offers_termination_reason_message( + instance_model: InstanceModel, + profile: Profile, + requirements: Requirements, + placement_group_model: Optional[PlacementGroupModel], + master_job_provisioning_data: Optional[JobProvisioningData], +) -> Optional[str]: + """Explain why no offers are available by refetching with unavailable offers included. + + Return "No quota" or "No balance" only if all matching offers have that reason. + Return None otherwise or if the lookup fails. + + Adds one offer lookup only when no available offers were found. Backend loading and + quota/balance checks may repeat DB reads; provider cache misses may repeat + provider calls. + """ + # TODO: Return InstanceTerminationReason values and persist NO_QUOTA/NO_BALANCE + # once all supported rolling-upgrade source versions recognize them. Until then, + # use messages so older replicas can read the shared DB. + # Refetching will still be needed: the initial lookup excludes unavailable offers. + try: + offers = await get_fleet_offers( + project=instance_model.project, + profile=profile, + requirements=requirements, + exclude_not_available=False, + fleet_model=instance_model.fleet, + placement_group=placement_group_model_to_placement_group_optional( + placement_group_model + ), + blocks="auto" if instance_model.total_blocks is None else instance_model.total_blocks, + master_job_provisioning_data=master_job_provisioning_data, + infer_master_job_provisioning_data_from_fleet_instances=False, + include_only_create_instance_supported_backends=True, + ) + return await run_async(get_offers_termination_reason_message, offers) + except Exception: + logger.warning( + "%s: failed to check unavailable offers", fmt(instance_model), exc_info=True + ) + return None + + async def _get_cluster_master_context( instance_model: InstanceModel, ) -> Optional[_ClusterMasterContext]: diff --git a/src/dstack/_internal/server/background/pipeline_tasks/jobs_submitted.py b/src/dstack/_internal/server/background/pipeline_tasks/jobs_submitted.py index 041574799d..2d93fb0d4f 100644 --- a/src/dstack/_internal/server/background/pipeline_tasks/jobs_submitted.py +++ b/src/dstack/_internal/server/background/pipeline_tasks/jobs_submitted.py @@ -31,7 +31,6 @@ InstanceGroupPlacement, ) from dstack._internal.core.models.instances import ( - InstanceAvailability, InstanceOfferWithAvailability, InstanceStatus, ) @@ -120,6 +119,7 @@ from dstack._internal.server.services.offers import ( get_instance_offer_with_restricted_az, get_offers_by_requirements, + get_offers_termination_reason_message, ) from dstack._internal.server.services.pipelines import PipelineHinterProtocol from dstack._internal.server.services.placement import ( @@ -2421,21 +2421,8 @@ async def _provision_new_capacity( instance_mounts=check_run_spec_requires_instance_mounts(run.run_spec), placement_group=placement_group_model_to_placement_group_optional(placement_group_model), ) - available_offers = offers.offers - for availability, message in ( - (InstanceAvailability.NO_QUOTA, "No quota"), - (InstanceAvailability.NO_BALANCE, "No balance"), - ): - if offers.unavailable_reason == availability: - return _FailedNewCapacityProvisioning( - placement_group_cleanup=None, - # TODO: Store FAILED_TO_START_DUE_TO_NO_QUOTA/NO_BALANCE once the minimum - # supported rolling-upgrade source version handles them on every replica. - message=message, - ) - extra_authorized_keys = get_extra_authorized_keys(run.run_spec) - offers_iter = iter(available_offers) + offers_iter = iter(offers) offers_tried = 0 offers_taken = 0 skip_reasons: list[str] = [] @@ -2583,18 +2570,34 @@ async def _provision_new_capacity( offer.region, ) continue - return _FailedNewCapacityProvisioning( - message=_get_new_capacity_failure_message( - fleet_name=fleet_model.name, - attempts=_NewCapacityAttempts( - total=len(available_offers), - tried=offers_tried, - skip_reasons=skip_reasons, - errors=offer_errors, - # Offers are left only if the attempt limit, not the offer list, ended the loop. - limit_reached=offers_taken < len(available_offers), - ), + termination_reason_message = _get_new_capacity_failure_message( + fleet_name=fleet_model.name, + attempts=_NewCapacityAttempts( + total=len(offers), + tried=offers_tried, + skip_reasons=skip_reasons, + errors=offer_errors, + # Offers are left only if the attempt limit, not the offer list, ended the loop. + limit_reached=offers_taken < len(offers), ), + ) + if not offers: + termination_reason_message = ( + await _get_no_offers_termination_reason_message( + project=project, + job_model=job_model, + run=run, + job=job, + profile=profile, + requirements=requirements, + volumes=volumes, + placement_group_model=placement_group_model, + master_job_provisioning_data=master_job_provisioning_data, + ) + or termination_reason_message + ) + return _FailedNewCapacityProvisioning( + message=termination_reason_message, placement_group_cleanup=_build_placement_group_cleanup( fleet_model=fleet_model, offers_tried=offers_tried, @@ -2604,6 +2607,52 @@ async def _provision_new_capacity( ) +async def _get_no_offers_termination_reason_message( + project: ProjectModel, + job_model: JobModel, + run: Run, + job: Job, + profile: Profile, + requirements: Requirements, + volumes: list[list[Volume]], + placement_group_model: Optional[PlacementGroupModel], + master_job_provisioning_data: Optional[JobProvisioningData], +) -> Optional[str]: + """Explain why no offers are available by refetching with unavailable offers included. + + Return "No quota" or "No balance" only if all matching offers have that reason. + Return None otherwise or if the lookup fails. + + Adds one offer lookup only when no available offers were found. Backend loading and + quota/balance checks may repeat DB reads; provider cache misses may repeat + provider calls. + """ + # TODO: Return JobTerminationReason values and persist + # FAILED_TO_START_DUE_TO_NO_QUOTA/NO_BALANCE once all supported rolling-upgrade + # source versions recognize them. Until then, use messages so older replicas + # can read the shared DB. + # Refetching will still be needed: the initial lookup excludes unavailable offers. + try: + offers = await get_offers_by_requirements( + project=project, + profile=profile, + requirements=requirements, + exclude_not_available=False, + multinode=requirements.multinode or is_multinode_job(job), + master_job_provisioning_data=master_job_provisioning_data, + volumes=volumes, + privileged=job.job_spec.privileged, + instance_mounts=check_run_spec_requires_instance_mounts(run.run_spec), + placement_group=placement_group_model_to_placement_group_optional( + placement_group_model + ), + ) + return await run_async(get_offers_termination_reason_message, offers) + except Exception: + logger.warning("%s: failed to check unavailable offers", fmt(job_model), exc_info=True) + return None + + _MAX_OFFER_ERROR_LEN = 200 diff --git a/src/dstack/_internal/server/services/backends/__init__.py b/src/dstack/_internal/server/services/backends/__init__.py index f15845b82c..8472375b7a 100644 --- a/src/dstack/_internal/server/services/backends/__init__.py +++ b/src/dstack/_internal/server/services/backends/__init__.py @@ -40,6 +40,7 @@ from dstack._internal.core.models.runs import Requirements from dstack._internal.server import settings from dstack._internal.server.models import BackendModel, DecryptedString, ProjectModel +from dstack._internal.server.services.offers import merge_offer_iterables from dstack._internal.utils.common import run_async from dstack._internal.utils.logging import get_logger @@ -471,17 +472,26 @@ async def get_backend_offers( requirements: Requirements, full_offers: bool, unallocated_resources: bool, -) -> list[tuple[Backend, Iterable[InstanceOfferWithAvailability]]]: + exclude_not_available: bool = False, +) -> Iterable[Tuple[Backend, InstanceOfferWithAvailability]]: """ - Return each backend's lazy offers satisfying `requirements`. + Returns a lazy iterable of backend offers satisfying `requirements` sorted by price. """ + def get_filtered_offers_with_backends( + backend: Backend, + offers: Iterable[InstanceOfferWithAvailability], + ) -> Iterator[Tuple[Backend, InstanceOfferWithAvailability]]: + for offer in offers: + if not exclude_not_available or offer.availability.is_available(): + yield (backend, offer) + logger.debug("Requesting instance offers from backends: %s", [b.TYPE.value for b in backends]) tasks = [ run_async(get_offers_tracked, backend, requirements, full_offers, unallocated_resources) for backend in backends ] - offers_by_backend: list[tuple[Backend, Iterable[InstanceOfferWithAvailability]]] = [] + offers_by_backend: list[Iterable[tuple[Backend, InstanceOfferWithAvailability]]] = [] for backend, result in zip(backends, await asyncio.gather(*tasks, return_exceptions=True)): if isinstance(result, BackendError): logger.warning( @@ -497,8 +507,8 @@ async def get_backend_offers( exc_info=result, ) continue - offers_by_backend.append((backend, result)) - return offers_by_backend + offers_by_backend.append(get_filtered_offers_with_backends(backend, result)) + return merge_offer_iterables(*offers_by_backend) def check_backend_type_available(backend_type: BackendType): diff --git a/src/dstack/_internal/server/services/fleets.py b/src/dstack/_internal/server/services/fleets.py index 6bdaa9b56f..ac677b4948 100644 --- a/src/dstack/_internal/server/services/fleets.py +++ b/src/dstack/_internal/server/services/fleets.py @@ -3,12 +3,13 @@ from collections.abc import Callable from datetime import datetime from functools import wraps -from typing import List, Literal, Optional, TypeVar, Union +from typing import List, Literal, Optional, Tuple, TypeVar, Union from sqlalchemy import and_, exists, false, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import aliased, joinedload, selectinload +from dstack._internal.core.backends.base.backend import Backend from dstack._internal.core.backends.features import BACKENDS_WITH_CREATE_INSTANCE_SUPPORT from dstack._internal.core.errors import ( ForbiddenError, @@ -31,6 +32,7 @@ SSHParams, ) from dstack._internal.core.models.instances import ( + InstanceOfferWithAvailability, InstanceStatus, InstanceTerminationReason, SSHConnectionParams, @@ -468,7 +470,7 @@ async def get_plan( include_only_create_instance_supported_backends ), ) - offers = [offer for _, offer in offers_with_backends.offers] + offers = [offer for _, offer in offers_with_backends] _remove_fleet_spec_sensitive_info(effective_spec) if current_fleet is not None: @@ -499,7 +501,7 @@ async def get_fleet_offers( master_job_provisioning_data: Optional[JobProvisioningData] = None, infer_master_job_provisioning_data_from_fleet_instances: bool = True, include_only_create_instance_supported_backends: bool = True, -) -> offers_services.Offers: +) -> List[Tuple[Backend, InstanceOfferWithAvailability]]: """ Return offers for fleet planning and provisioning. @@ -534,10 +536,13 @@ async def get_fleet_offers( master_job_provisioning_data=master_job_provisioning_data, placement_group=placement_group, blocks=blocks, - supported_backend_types=BACKENDS_WITH_CREATE_INSTANCE_SUPPORT - if include_only_create_instance_supported_backends - else None, ) + if include_only_create_instance_supported_backends: + offers = [ + (backend, offer) + for backend, offer in offers + if offer.backend in BACKENDS_WITH_CREATE_INSTANCE_SUPPORT + ] return offers diff --git a/src/dstack/_internal/server/services/jobs/__init__.py b/src/dstack/_internal/server/services/jobs/__init__.py index 5a15e8403f..1ebbeecf9b 100644 --- a/src/dstack/_internal/server/services/jobs/__init__.py +++ b/src/dstack/_internal/server/services/jobs/__init__.py @@ -9,7 +9,11 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import load_only -from dstack._internal.core.consts import DSTACK_RUNNER_HTTP_PORT +from dstack._internal.core.consts import ( + DSTACK_RUNNER_HTTP_PORT, + NO_BALANCE_MESSAGE, + NO_QUOTA_MESSAGE, +) from dstack._internal.core.errors import ( ResourceNotExistsError, ServerClientError, @@ -625,7 +629,7 @@ def _get_job_status_message(job_model: JobModel) -> str: and "No matching fleet found" in job_model.termination_reason_message ): return "no fleets" - if job_model.termination_reason_message in ("No quota", "No balance"): + if job_model.termination_reason_message in (NO_QUOTA_MESSAGE, NO_BALANCE_MESSAGE): return job_model.termination_reason_message.lower() return "no capacity" elif job_model.termination_reason == JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA: diff --git a/src/dstack/_internal/server/services/offers.py b/src/dstack/_internal/server/services/offers.py index 5dea8d665a..1e1342ff6f 100644 --- a/src/dstack/_internal/server/services/offers.py +++ b/src/dstack/_internal/server/services/offers.py @@ -1,7 +1,6 @@ import heapq import itertools from collections.abc import Container, Iterable, Iterator -from dataclasses import dataclass from typing import List, Literal, Optional, Tuple, TypeVar, Union import gpuhunt @@ -14,6 +13,7 @@ BACKENDS_WITH_PRIVILEGED_SUPPORT, BACKENDS_WITH_RESERVATION_SUPPORT, ) +from dstack._internal.core.consts import NO_BALANCE_MESSAGE, NO_QUOTA_MESSAGE from dstack._internal.core.models.backends.base import BackendType from dstack._internal.core.models.instances import ( InstanceAvailability, @@ -30,18 +30,11 @@ from dstack._internal.utils.common import run_async -@dataclass -class Offers: - offers: list[tuple[Backend, InstanceOfferWithAvailability]] - unavailable_reason: Optional[InstanceAvailability] = None - """Common reason for excluded matching offers if none are available; None otherwise.""" - - async def get_offers_by_requirements( project: ProjectModel, profile: Profile, requirements: Requirements, - exclude_not_available: bool = False, + exclude_not_available=False, multinode: bool = False, master_job_provisioning_data: Optional[JobProvisioningData] = None, volumes: Optional[List[List[Volume]]] = None, @@ -52,8 +45,7 @@ async def get_offers_by_requirements( max_offers: Optional[int] = None, full_offers: bool = False, unallocated_resources: bool = False, - supported_backend_types: Optional[Container[BackendType]] = None, -) -> Offers: +) -> List[Tuple[Backend, InstanceOfferWithAvailability]]: backends: List[Backend] = await backends_services.get_project_backends(project=project) backend_types: Optional[list[BackendType]] = profile.backends @@ -109,60 +101,24 @@ async def get_offers_by_requirements( if backend_types is not None: backends = [b for b in backends if b.TYPE in backend_types or b.TYPE == BackendType.DSTACK] - offer_filter = _OfferFilter( - backend_types=backend_types, - regions=None if regions is None else [r.lower() for r in regions], - availability_zones=None - if availability_zones is None - else [z.lower() for z in availability_zones], - instance_types=None if instance_types is None else [i.lower() for i in instance_types], - placement_group=placement_group, - volumes_locations=volumes_locations, - supported_backend_types=supported_backend_types, - ) - - offers_by_backend = await backends_services.get_backend_offers( + offers = await backends_services.get_backend_offers( backends=backends, requirements=requirements, full_offers=full_offers, unallocated_resources=unallocated_resources, + exclude_not_available=exclude_not_available, ) - return await run_async( - _select_offers, offers_by_backend, offer_filter, blocks, exclude_not_available, max_offers - ) - -def _select_offers( - offers_by_backend: list[tuple[Backend, Iterable[InstanceOfferWithAvailability]]], - offer_filter: "_OfferFilter", - blocks: Union[int, Literal["auto"]], - exclude_not_available: bool, - max_offers: Optional[int], -) -> Offers: - unavailable: set[InstanceAvailability] = set() - - def available_offers( - backend: Backend, backend_offers: Iterable[InstanceOfferWithAvailability] - ) -> Iterator[tuple[Backend, InstanceOfferWithAvailability]]: - for offer in backend_offers: - if offer.availability.is_available(): - yield backend, offer - elif ( - offer.availability not in unavailable - and offer_filter.matches(backend, offer) - and (blocks == 1 or _get_offer_blocks(offer, blocks)[0]) - ): - unavailable.add(offer.availability) - - iterables = [ - available_offers(backend, backend_offers) - if exclude_not_available - else _with_backend(backend, backend_offers) - for backend, backend_offers in offers_by_backend - ] - - # Availability must be filtered before merging to preserve backend-specific order. - offers = _filter_offers(merge_offer_iterables(*iterables), offer_filter) + offers = _filter_offers( + offers=offers, + # Double filtering by backends if backend returns offers for other backend. + backend_types=backend_types, + regions=regions, + availability_zones=availability_zones, + instance_types=instance_types, + placement_group=placement_group, + volumes_locations=volumes_locations, + ) if blocks != 1: offers = _get_shareable_offers(offers, blocks) @@ -170,21 +126,29 @@ def available_offers( if max_offers is not None: offers = itertools.islice(offers, max_offers) + # Sorting consumes the lazy offers chain, which is CPU-bound, + # so it must not run on the event loop. + return await run_async(_sort_offers_by_availability, offers) + + +def get_offers_termination_reason_message( + offers: Iterable[Tuple[Backend, InstanceOfferWithAvailability]], +) -> Optional[str]: + availabilities = {offer.availability for _, offer in offers} + if availabilities == {InstanceAvailability.NO_QUOTA}: + return NO_QUOTA_MESSAGE + if availabilities == {InstanceAvailability.NO_BALANCE}: + return NO_BALANCE_MESSAGE + return None + + +def _sort_offers_by_availability( + offers: Iterable[Tuple[Backend, InstanceOfferWithAvailability]], +) -> List[Tuple[Backend, InstanceOfferWithAvailability]]: # Put NOT_AVAILABLE and NO_QUOTA offers at the end. # We have to do this after taking max_offers to avoid processing all offers # if all/most offers are unavailable. - selected = sorted(offers, key=lambda i: not i[1].availability.is_available()) - unavailable_reason = ( - next(iter(unavailable)) if not selected and len(unavailable) == 1 else None - ) - return Offers(offers=selected, unavailable_reason=unavailable_reason) - - -def _with_backend( - backend: Backend, offers: Iterable[InstanceOfferWithAvailability] -) -> Iterator[tuple[Backend, InstanceOfferWithAvailability]]: - for offer in offers: - yield backend, offer + return sorted(offers, key=lambda i: not i[1].availability.is_available()) T = TypeVar("T") @@ -265,81 +229,56 @@ def get_instance_offer_with_restricted_az( return instance_offer -@dataclass -class _OfferFilter: - backend_types: Optional[List[BackendType]] - regions: Optional[List[str]] - availability_zones: Optional[List[str]] - instance_types: Optional[List[str]] - placement_group: Optional[PlacementGroup] - volumes_locations: Optional[Container[tuple[BackendType, Optional[str]]]] - supported_backend_types: Optional[Container[BackendType]] - - def matches( - self, - backend: Backend, - offer: InstanceOfferWithAvailability, - check_availability_zones: bool = True, - ) -> bool: - # Backends such as dstack can return offers for other backend types. - if self.backend_types is not None and offer.backend not in self.backend_types: - return False - if ( - self.supported_backend_types is not None - and offer.backend not in self.supported_backend_types - ): - return False - if self.regions is not None and offer.region.lower() not in self.regions: - return False - if ( - self.instance_types is not None - and offer.instance.name.lower() not in self.instance_types - ): - return False - if self.placement_group is not None: - compute = backend.compute() - if not isinstance( - compute, ComputeWithPlacementGroupSupport - ) or not compute.is_suitable_placement_group(self.placement_group, offer): - return False - if ( - check_availability_zones - and self.availability_zones is not None - and ( - offer.availability_zones is None - or not any(z.lower() in self.availability_zones for z in offer.availability_zones) - ) - ): - return False - # Offer is futher filtered against volumes AZs in Compute implementation, see - # ComputeWithCreateInstanceSupport._restrict_instance_offer_az_to_volumes_az() - if ( - self.volumes_locations is not None - and (offer.backend, offer.region.lower()) not in self.volumes_locations - and (offer.backend, None) not in self.volumes_locations - ): - return False - return True - - def _filter_offers( offers: Iterable[Tuple[Backend, InstanceOfferWithAvailability]], - offer_filter: _OfferFilter, + backend_types: Optional[List[BackendType]] = None, + regions: Optional[List[str]] = None, + availability_zones: Optional[List[str]] = None, + instance_types: Optional[List[str]] = None, + placement_group: Optional[PlacementGroup] = None, + volumes_locations: Optional[Container[tuple[BackendType, Optional[str]]]] = None, ) -> Iterator[Tuple[Backend, InstanceOfferWithAvailability]]: + """ + Yields filtered offers. May return modified offers to match the filters. + """ + if regions is not None: + regions = [r.lower() for r in regions] + if instance_types is not None: + instance_types = [i.lower() for i in instance_types] + if availability_zones is not None: + availability_zones = [z.lower() for z in availability_zones] + for b, offer in offers: - # Check AZs once while preparing the restricted offer below. - if not offer_filter.matches(b, offer, check_availability_zones=False): + if backend_types is not None and offer.backend not in backend_types: continue - if offer_filter.availability_zones is not None: + if regions is not None and offer.region.lower() not in regions: + continue + if instance_types is not None and offer.instance.name.lower() not in instance_types: + continue + if placement_group is not None: + compute = b.compute() + if not isinstance( + compute, ComputeWithPlacementGroupSupport + ) or not compute.is_suitable_placement_group(placement_group, offer): + continue + if availability_zones is not None: if offer.availability_zones is None: continue new_offer = offer.model_copy() new_offer.availability_zones = [ - z for z in offer.availability_zones if z.lower() in offer_filter.availability_zones + z for z in offer.availability_zones if z.lower() in availability_zones ] if not new_offer.availability_zones: continue offer = new_offer + # Offer is futher filtered against volumes AZs in Compute implementation, see + # ComputeWithCreateInstanceSupport._restrict_instance_offer_az_to_volumes_az() + if ( + volumes_locations is not None + and (offer.backend, offer.region.lower()) not in volumes_locations + and (offer.backend, None) not in volumes_locations + ): + continue yield (b, offer) @@ -351,20 +290,15 @@ def _get_shareable_offers( Yields offers that can be shared with `total_blocks` set. """ for backend, offer in offers: - divisible, total_blocks = _get_offer_blocks(offer, blocks) + resources = offer.instance.resources + cpu_count = resources.cpus + gpu_count = len(resources.gpus) + if gpu_count > 0 and resources.gpus[0].vendor == gpuhunt.AcceleratorVendor.GOOGLE: + # TPUs cannot be shared + gpu_count = 1 + divisible, total_blocks = is_divisible_into_blocks(cpu_count, gpu_count, blocks) if not divisible: continue new_offer = offer.model_copy() new_offer.total_blocks = total_blocks yield (backend, new_offer) - - -def _get_offer_blocks( - offer: InstanceOfferWithAvailability, blocks: Union[int, Literal["auto"]] -) -> tuple[bool, int]: - resources = offer.instance.resources - gpu_count = len(resources.gpus) - if gpu_count > 0 and resources.gpus[0].vendor == gpuhunt.AcceleratorVendor.GOOGLE: - # TPUs cannot be shared - gpu_count = 1 - return is_divisible_into_blocks(resources.cpus, gpu_count, blocks) diff --git a/src/dstack/_internal/server/services/runs/plan.py b/src/dstack/_internal/server/services/runs/plan.py index 4b5e3e5895..c5c175c57b 100644 --- a/src/dstack/_internal/server/services/runs/plan.py +++ b/src/dstack/_internal/server/services/runs/plan.py @@ -794,7 +794,7 @@ async def _get_backend_offers_in_fleet( # Handle multinode for old jobs that don't have requirements.multinode set. # TODO: Drop multinode param. multinode = requirements.multinode or is_multinode_job(job) - offers_result = await get_offers_by_requirements( + backend_offers = await get_offers_by_requirements( project=project, profile=profile, requirements=requirements, @@ -807,7 +807,6 @@ async def _get_backend_offers_in_fleet( full_offers=full_offers, unallocated_resources=unallocated_resources, ) - backend_offers = offers_result.offers return backend_offers @@ -875,7 +874,7 @@ async def get_non_fleet_offers( if skip_backend_offers: backend_offers = [] else: - offers_result = await get_offers_by_requirements( + backend_offers = await get_offers_by_requirements( project=project, profile=run_spec.merged_profile, requirements=job.job_spec.requirements, @@ -887,7 +886,6 @@ async def get_non_fleet_offers( full_offers=full_offers, unallocated_resources=unallocated_resources, ) - backend_offers = offers_result.offers return instance_offers, backend_offers diff --git a/src/tests/_internal/cli/utils/test_run.py b/src/tests/_internal/cli/utils/test_run.py index 0d643dab81..2f1044bab7 100644 --- a/src/tests/_internal/cli/utils/test_run.py +++ b/src/tests/_internal/cli/utils/test_run.py @@ -11,7 +11,6 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload -from dstack._internal.cli.services.configurators.run import print_finished_message from dstack._internal.cli.utils.run import _format_pull_progress, get_runs_table from dstack._internal.core.models.backends.base import BackendType from dstack._internal.core.models.configurations import ( @@ -596,34 +595,3 @@ async def test_service_with_multiple_replicas_and_jobs(self, session: AsyncSessi ) def test_format_pull_progress(progress: ImagePullProgress, expected: str) -> None: assert _format_pull_progress(progress) == expected - - -@pytest.mark.asyncio -@pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True) -@pytest.mark.usefixtures("image_config_mock") -class TestPrintFinishedMessage: - @pytest.mark.parametrize( - ("reason", "status_message"), - [ - (JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, "No capacity"), - (JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA, "No quota"), - (JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE, "No balance"), - ], - ) - async def test_provisioning_failure_has_no_runner_logs_hint( - self, - test_db, - session: AsyncSession, - capsys: pytest.CaptureFixture[str], - reason: JobTerminationReason, - status_message: str, - ): - run = await create_run_with_job( - session=session, job_status=JobStatus.FAILED, termination_reason=reason - ) - - print_finished_message(run) - - output = capsys.readouterr().out - assert status_message in output - assert "dstack logs" not in output diff --git a/src/tests/_internal/core/models/test_runs.py b/src/tests/_internal/core/models/test_runs.py index 105d6d86b8..170f450d3b 100644 --- a/src/tests/_internal/core/models/test_runs.py +++ b/src/tests/_internal/core/models/test_runs.py @@ -78,19 +78,6 @@ def test_job_termination_reason_to_retry_event_works_with_all_enum_variants(): assert retry_event is None or isinstance(retry_event, RetryEvent) -class TestJobTerminationReason: - @pytest.mark.parametrize( - "reason", - [ - JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA, - JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE, - ], - ) - def test_blocked_offers_preserve_no_capacity_retry_policy(self, reason): - assert reason.to_status() == JobStatus.FAILED - assert reason.to_retry_event() == RetryEvent.NO_CAPACITY - - # Will fail if JobTerminationReason value is added without updating JobSubmission._get_error def test_get_error_returns_expected_messages(): # already handled and shown in status_message diff --git a/src/tests/_internal/server/background/pipeline_tasks/test_instances/test_cloud_provisioning.py b/src/tests/_internal/server/background/pipeline_tasks/test_instances/test_cloud_provisioning.py index 4d0ec29f4c..b8c8213c16 100644 --- a/src/tests/_internal/server/background/pipeline_tasks/test_instances/test_cloud_provisioning.py +++ b/src/tests/_internal/server/background/pipeline_tasks/test_instances/test_cloud_provisioning.py @@ -108,10 +108,7 @@ async def test_creates_instance( total_blocks=expected_blocks, ) backend_mock.compute.return_value = Mock(spec=ComputeMockSpec) - backend_mock.compute.return_value.get_offers.return_value = [ - offer.model_copy(update={"availability": InstanceAvailability.NO_QUOTA}), - offer, - ] + backend_mock.compute.return_value.get_offers.return_value = [offer] backend_mock.compute.return_value.create_instance.return_value = JobProvisioningData( backend=offer.backend, instance_type=offer.instance, @@ -133,7 +130,7 @@ async def test_creates_instance( assert instance.status == InstanceStatus.PROVISIONING assert instance.total_blocks == expected_blocks assert instance.busy_blocks == 0 - backend_mock.compute.return_value.create_instance.assert_called_once() + backend_mock.compute.return_value.get_offers.assert_called_once() @pytest.mark.parametrize("err", [RuntimeError("Unexpected"), ProvisioningError("Expected")]) async def test_tries_second_offer_if_first_fails( @@ -201,6 +198,8 @@ async def test_fails_if_all_offers_fail( await session.refresh(instance) assert instance.status == InstanceStatus.TERMINATED assert instance.termination_reason == InstanceTerminationReason.NO_OFFERS + assert instance.termination_reason_message == "All offers failed" + aws_mock.compute.return_value.get_offers.assert_called_once() async def test_fails_if_no_offers( self, @@ -221,6 +220,7 @@ async def test_fails_if_no_offers( await session.refresh(instance) assert instance.status == InstanceStatus.TERMINATED assert instance.termination_reason == InstanceTerminationReason.NO_OFFERS + assert instance.termination_reason_message == "No offers found" @pytest.mark.parametrize( ("availabilities", "message"), @@ -231,14 +231,6 @@ async def test_fails_if_no_offers( "No offers found", ), ([InstanceAvailability.NO_BALANCE] * 2, "No balance"), - ( - [InstanceAvailability.NO_QUOTA, InstanceAvailability.NO_BALANCE], - "No offers found", - ), - ( - [InstanceAvailability.NO_BALANCE, InstanceAvailability.NOT_AVAILABLE], - "No offers found", - ), ], ) async def test_fails_without_attempting_unavailable_offers( @@ -270,7 +262,7 @@ async def test_fails_without_attempting_unavailable_offers( assert instance.status == InstanceStatus.TERMINATED assert instance.termination_reason == InstanceTerminationReason.NO_OFFERS assert instance.termination_reason_message == message - compute.get_offers.assert_called_once() + assert compute.get_offers.call_count == 2 compute.create_instance.assert_not_called() async def test_waits_when_fleet_has_no_current_master( diff --git a/src/tests/_internal/server/background/pipeline_tasks/test_runs/test_active.py b/src/tests/_internal/server/background/pipeline_tasks/test_runs/test_active.py index add2e30487..b2c2971725 100644 --- a/src/tests/_internal/server/background/pipeline_tasks/test_runs/test_active.py +++ b/src/tests/_internal/server/background/pipeline_tasks/test_runs/test_active.py @@ -558,16 +558,8 @@ async def test_retries_no_capacity_replica_and_keeps_service_running( assert retried_job.status == JobStatus.SUBMITTED assert len(jobs) == 3 - @pytest.mark.parametrize( - "reason", - [ - JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, - JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA, - JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE, - ], - ) async def test_replica_retry_deletes_superseded_no_capacity_submissions( - self, test_db, session: AsyncSession, worker: RunWorker, reason: JobTerminationReason + self, test_db, session: AsyncSession, worker: RunWorker ) -> None: project = await create_project(session=session) user = await create_user(session=session) @@ -605,7 +597,7 @@ async def test_replica_retry_deletes_superseded_no_capacity_submissions( session=session, run=run, status=JobStatus.FAILED, - termination_reason=reason, + termination_reason=JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, submitted_at=run.submitted_at, last_processed_at=run.last_processed_at, replica_num=0, @@ -646,16 +638,8 @@ async def test_replica_retry_deletes_superseded_no_capacity_submissions( assert interrupted_job.id in [j.id for j in jobs] assert jobs[-1].status == JobStatus.SUBMITTED - @pytest.mark.parametrize( - "reason", - [ - JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, - JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA, - JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE, - ], - ) async def test_retries_scheduled_run_no_capacity_from_trigger_time( - self, test_db, session: AsyncSession, worker: RunWorker, reason: JobTerminationReason + self, test_db, session: AsyncSession, worker: RunWorker ) -> None: project = await create_project(session=session) user = await create_user(session=session) @@ -687,7 +671,7 @@ async def test_retries_scheduled_run_no_capacity_from_trigger_time( session=session, run=run, status=JobStatus.FAILED, - termination_reason=reason, + termination_reason=JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY, ) lock_run(run) await session.commit() diff --git a/src/tests/_internal/server/background/pipeline_tasks/test_submitted_jobs.py b/src/tests/_internal/server/background/pipeline_tasks/test_submitted_jobs.py index befd73ef86..ff83046e09 100644 --- a/src/tests/_internal/server/background/pipeline_tasks/test_submitted_jobs.py +++ b/src/tests/_internal/server/background/pipeline_tasks/test_submitted_jobs.py @@ -6,7 +6,7 @@ import gpuhunt import pytest -from sqlalchemy import event, select +from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import joinedload @@ -431,10 +431,7 @@ async def test_provisions_new_capacity_for_assigned_job_with_placeholder( backend_mock = Mock() m.return_value = [backend_mock] backend_mock.TYPE = BackendType.AWS - backend_mock.compute.return_value.get_offers.return_value = [ - offer.model_copy(update={"availability": InstanceAvailability.NO_QUOTA}), - offer, - ] + backend_mock.compute.return_value.get_offers.return_value = [offer] backend_mock.compute.return_value.run_job.return_value = get_job_provisioning_data( dockerized=True, backend=BackendType.AWS, @@ -461,7 +458,6 @@ async def test_provisions_new_capacity_for_assigned_job_with_placeholder( assert job.instance.fleet_id == fleet.id assert job.instance.offer is not None assert job.instance.provisioning_job_id == job.id # never cleared - backend_mock.compute.return_value.run_job.assert_called_once() res = await session.execute( select(InstanceModel).where( InstanceModel.fleet_id == fleet.id, @@ -2041,15 +2037,8 @@ async def test_leaves_placeholder_for_terminating_pipeline_on_failed_new_capacit assert not placeholder.deleted assert placeholder.status == InstanceStatus.PENDING - @pytest.mark.parametrize( - "blocked_availability", [InstanceAvailability.NO_QUOTA, InstanceAvailability.NO_BALANCE] - ) async def test_reports_tried_offers_when_new_capacity_provisioning_fails( - self, - test_db, - session: AsyncSession, - worker: JobSubmittedWorker, - blocked_availability: InstanceAvailability, + self, test_db, session: AsyncSession, worker: JobSubmittedWorker ): project = await create_project(session=session) user = await create_user(session=session) @@ -2067,14 +2056,12 @@ async def test_reports_tried_offers_when_new_capacity_provisioning_fails( backend_mock.TYPE = BackendType.AWS backend_mock.compute.return_value = compute_mock m.return_value = [backend_mock] - compute_mock.get_offers.return_value = [ - offer.model_copy(update={"availability": blocked_availability}), - offer, - ] + compute_mock.get_offers.return_value = [offer] compute_mock.run_job.side_effect = BackendError("InsufficientInstanceCapacity") # The first pass assigns the job to the fleet, the second one provisions. await _process_job(session=session, worker=worker, job_model=job) + compute_mock.get_offers.reset_mock() await _process_job(session=session, worker=worker, job_model=job) job = await _get_job(session, job.id) @@ -2083,7 +2070,7 @@ async def test_reports_tried_offers_when_new_capacity_provisioning_fails( assert f"Failed to provision in fleet '{fleet.name}'" in job.termination_reason_message assert "tried 1 of 1 offers, all failed" in job.termination_reason_message assert "us-east-1: InsufficientInstanceCapacity" in job.termination_reason_message - compute_mock.run_job.assert_called_once() + compute_mock.get_offers.assert_called_once() @pytest.mark.parametrize( ("availabilities", "message"), @@ -2101,14 +2088,6 @@ async def test_reports_tried_offers_when_new_capacity_provisioning_fails( [InstanceAvailability.NO_BALANCE], "No balance", ), - ( - [InstanceAvailability.NO_QUOTA, InstanceAvailability.NO_BALANCE], - None, - ), - ( - [InstanceAvailability.NO_BALANCE, InstanceAvailability.NOT_AVAILABLE], - None, - ), ], ) async def test_fails_without_attempting_unavailable_offers( @@ -2147,87 +2126,43 @@ async def test_fails_without_attempting_unavailable_offers( else: assert job.termination_reason_message is not None assert "No offers matching the run requirements" in job.termination_reason_message - assert job.lock_token is None - assert job.instance is not None - assert not job.instance.deleted - assert job.instance.status == InstanceStatus.PENDING - compute.get_offers.assert_called_once() + assert compute.get_offers.call_count == 2 compute.run_job.assert_not_called() compute.run_jobs.assert_not_called() compute.create_instance.assert_not_called() - async def test_blocked_offers_keep_query_count_and_offer_iteration_outside_sessions( + async def test_keeps_generic_failure_if_refetch_fails( self, test_db, session: AsyncSession, worker: JobSubmittedWorker ): + project = await create_project(session=session) user = await create_user(session=session) - query_counts = [] - for index, availability in enumerate( - [None, InstanceAvailability.NO_QUOTA, InstanceAvailability.NO_BALANCE] + repo = await create_repo(session=session, project_id=project.id) + fleet = await create_fleet(session=session, project=project) + run = await create_run(session=session, project=project, repo=repo, user=user, fleet=fleet) + job = await create_job(session=session, run=run) + backend = Mock(TYPE=BackendType.DSTACK) + compute = Mock(spec=ComputeMockSpec) + backend.compute.return_value = compute + + compute.get_offers.return_value = [ + get_instance_offer_with_availability(availability=InstanceAvailability.NO_QUOTA) + ] + with patch( + "dstack._internal.server.services.backends.get_project_backends", + side_effect=[[backend], RuntimeError("Diagnostic lookup failed")], ): - project = await create_project(session=session, name=f"project_{index}", owner=user) - repo = await create_repo(session=session, project_id=project.id) - fleet = await create_fleet(session=session, project=project) - run = await create_run( - session=session, project=project, repo=repo, user=user, fleet=fleet - ) - job = await create_job(session=session, run=run) - await session.commit() - active_connections = 0 - queries = 0 - - def checkout(*args): - nonlocal active_connections - active_connections += 1 - - def checkin(*args): - nonlocal active_connections - active_connections -= 1 - - def before_cursor_execute(*args): - nonlocal queries - queries += 1 - - def iterate_offers(): - if availability is None: - return - offer = get_instance_offer_with_availability(availability=availability) - for _ in range(1000): - # The generator resumes after each offer's rejection is processed. - assert active_connections == 0 - previous_queries = queries - yield offer - assert queries == previous_queries - assert active_connections == 0 - - backend = Mock(TYPE=BackendType.DSTACK) - backend.compute.return_value.get_offers.side_effect = lambda *args: iterate_offers() - engine = test_db.engine.sync_engine - listeners = [ - ("checkout", checkout), - ("checkin", checkin), - ("before_cursor_execute", before_cursor_execute), - ] - for name, listener in listeners: - event.listen(engine, name, listener) - try: - with patch( - "dstack._internal.server.services.backends.get_project_backends", - return_value=[backend], - ): - await _process_job(session=session, worker=worker, job_model=job) - await _process_job(session=session, worker=worker, job_model=job) - query_counts.append(queries) - assert active_connections == 0 - finally: - for name, listener in listeners: - event.remove(engine, name, listener) - job = await _get_job(session, job.id) - assert ( - job.termination_reason == JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY - ) + await _process_job(session=session, worker=worker, job_model=job) + await _process_job(session=session, worker=worker, job_model=job) - assert query_counts[0] > 0 - assert len(set(query_counts)) == 1 + job = await _get_job(session, job.id) + assert job.status == JobStatus.TERMINATING + assert job.termination_reason == JobTerminationReason.FAILED_TO_START_DUE_TO_NO_CAPACITY + assert job.termination_reason_message is not None + assert "No offers matching the run requirements" in job.termination_reason_message + compute.get_offers.assert_called_once() + compute.run_job.assert_not_called() + compute.run_jobs.assert_not_called() + compute.create_instance.assert_not_called() async def test_provisions_compute_group( self, test_db, session: AsyncSession, worker: JobSubmittedWorker diff --git a/src/tests/_internal/server/services/jobs/test_jobs.py b/src/tests/_internal/server/services/jobs/test_jobs.py index 9efddbc73d..93a083f1a0 100644 --- a/src/tests/_internal/server/services/jobs/test_jobs.py +++ b/src/tests/_internal/server/services/jobs/test_jobs.py @@ -29,20 +29,6 @@ class TestGetJobStatusMessage: - @pytest.mark.parametrize( - ("reason", "message"), - [ - (JobTerminationReason.FAILED_TO_START_DUE_TO_NO_QUOTA, "no quota"), - (JobTerminationReason.FAILED_TO_START_DUE_TO_NO_BALANCE, "no balance"), - ], - ) - def test_reports_blocked_offers(self, reason, message): - job = JobModel( - status=JobStatus.FAILED, - termination_reason=reason, - ) - assert _get_job_status_message(job) == message - @pytest.mark.parametrize( ("message", "expected"), [("No quota", "no quota"), ("No balance", "no balance"), ("Other error", "no capacity")], diff --git a/src/tests/_internal/server/services/runs/test_plan.py b/src/tests/_internal/server/services/runs/test_plan.py index e943eb40df..e55ff6de67 100644 --- a/src/tests/_internal/server/services/runs/test_plan.py +++ b/src/tests/_internal/server/services/runs/test_plan.py @@ -24,7 +24,6 @@ ) from dstack._internal.core.models.resources import CPUSpec, GPUSpec, Memory, Range, ResourcesSpec from dstack._internal.server.services.jobs import get_jobs_from_run_spec -from dstack._internal.server.services.offers import Offers from dstack._internal.server.services.projects import get_project_model_by_name from dstack._internal.server.services.runs import get_plan from dstack._internal.server.services.runs.plan import ( @@ -838,7 +837,7 @@ async def test_keeps_unconstrained_offers_for_non_empty_cluster_fleet_without_el ) offer = get_instance_offer_with_availability() backend = AsyncMock() - get_offers_by_requirements_mock.return_value = Offers(offers=[(backend, offer)]) + get_offers_by_requirements_mock.return_value = [(backend, offer)] offers = await _get_backend_offers_in_fleet( project=project, diff --git a/src/tests/_internal/server/services/test_offers.py b/src/tests/_internal/server/services/test_offers.py index fc6bdeac2f..7ff0b4fe4a 100644 --- a/src/tests/_internal/server/services/test_offers.py +++ b/src/tests/_internal/server/services/test_offers.py @@ -3,296 +3,20 @@ import pytest from dstack._internal.core.models.backends.base import BackendType -from dstack._internal.core.models.instances import ( - InstanceAvailability, - InstanceOfferWithAvailability, -) -from dstack._internal.core.models.placement import PlacementGroup from dstack._internal.core.models.profiles import Profile from dstack._internal.core.models.resources import ResourcesSpec from dstack._internal.core.models.runs import Requirements from dstack._internal.core.models.volumes import DaytonaVolumeConfiguration from dstack._internal.server.services.offers import get_offers_by_requirements from dstack._internal.server.testing.common import ( - ComputeMockSpec, get_instance_offer_with_availability, get_kubernetes_volume_configuration, - get_placement_group_configuration, get_volume, get_volume_configuration, ) class TestGetOffersByRequirements: - @pytest.mark.asyncio - @pytest.mark.parametrize("exclude_not_available", [False, True]) - async def test_stops_at_max_offers(self, exclude_not_available): - offer = get_instance_offer_with_availability() - backend = Mock(TYPE=BackendType.AWS) - - def iterate_offers(): - yield offer - raise AssertionError("Planning consumed offers beyond max_offers") - - backend.compute.return_value.get_offers.return_value = iterate_offers() - with patch( - "dstack._internal.server.services.backends.get_project_backends", - return_value=[backend], - ): - result = await get_offers_by_requirements( - project=Mock(), - profile=Profile(), - requirements=Requirements(resources=ResourcesSpec()), - max_offers=1, - exclude_not_available=exclude_not_available, - ) - assert result.offers == [(backend, offer)] - - @pytest.mark.asyncio - @pytest.mark.parametrize( - ("availabilities", "no_quota", "no_balance"), - [ - ([], False, False), - ([InstanceAvailability.NO_QUOTA], True, False), - ([InstanceAvailability.NO_BALANCE], False, True), - ([InstanceAvailability.NO_QUOTA, InstanceAvailability.NOT_AVAILABLE], False, False), - ([InstanceAvailability.NO_BALANCE, InstanceAvailability.NOT_AVAILABLE], False, False), - ([InstanceAvailability.NO_QUOTA, InstanceAvailability.NO_BALANCE], False, False), - ([InstanceAvailability.AVAILABLE, InstanceAvailability.NO_QUOTA], False, False), - ([InstanceAvailability.AVAILABLE, InstanceAvailability.NO_BALANCE], False, False), - ], - ) - async def test_collects_unavailable_offers_without_returning_them( - self, availabilities, no_quota, no_balance - ): - backend = Mock(TYPE=BackendType.DSTACK) - offers = [get_instance_offer_with_availability(availability=a) for a in availabilities] - backend.compute.return_value.get_offers.return_value = iter(offers) - with patch( - "dstack._internal.server.services.backends.get_project_backends", - return_value=[backend], - ): - result = await get_offers_by_requirements( - project=Mock(), - profile=Profile(), - requirements=Requirements(resources=ResourcesSpec()), - exclude_not_available=True, - ) - assert result.offers == [ - (backend, offer) for offer in offers if offer.availability.is_available() - ] - assert (result.unavailable_reason == InstanceAvailability.NO_QUOTA) == no_quota - assert (result.unavailable_reason == InstanceAvailability.NO_BALANCE) == no_balance - backend.compute.return_value.get_offers.assert_called_once() - - @pytest.mark.asyncio - @pytest.mark.parametrize( - ("profile", "ignored_offer", "blocks", "supported_backends"), - [ - ( - Profile(backends=[BackendType.AWS]), - get_instance_offer_with_availability(backend=BackendType.GCP), - 1, - None, - ), - ( - Profile(regions=["eu-west"]), - get_instance_offer_with_availability(region="us-east"), - 1, - None, - ), - ( - Profile(instance_types=["instance"]), - get_instance_offer_with_availability(instance_type="other"), - 1, - None, - ), - ( - Profile(availability_zones=["AZ1"]), - get_instance_offer_with_availability(availability_zones=["az2"]), - 1, - None, - ), - ( - Profile(availability_zones=["az1"]), - get_instance_offer_with_availability(availability_zones=None), - 1, - None, - ), - (Profile(), get_instance_offer_with_availability(cpu_count=3), 2, None), - ( - Profile(), - get_instance_offer_with_availability(backend=BackendType.KUBERNETES), - 1, - [BackendType.AWS], - ), - ], - ) - async def test_ignores_unavailable_offers_that_do_not_match( - self, - profile: Profile, - ignored_offer: InstanceOfferWithAvailability, - blocks, - supported_backends, - ): - quota_offer = get_instance_offer_with_availability( - availability=InstanceAvailability.NO_QUOTA, availability_zones=["az1"] - ) - ignored_offer = ignored_offer.model_copy( - update={"availability": InstanceAvailability.NOT_AVAILABLE} - ) - backend = Mock(TYPE=BackendType.DSTACK) - backend.compute.return_value.get_offers.return_value = [quota_offer, ignored_offer] - with patch( - "dstack._internal.server.services.backends.get_project_backends", - return_value=[backend], - ): - result = await get_offers_by_requirements( - project=Mock(), - profile=profile, - requirements=Requirements(resources=ResourcesSpec()), - exclude_not_available=True, - blocks=blocks, - supported_backend_types=supported_backends, - ) - assert result.offers == [] - assert result.unavailable_reason == InstanceAvailability.NO_QUOTA - - @pytest.mark.asyncio - async def test_does_not_copy_unavailable_offers_for_az_or_blocks(self): - offer = get_instance_offer_with_availability( - availability=InstanceAvailability.NO_QUOTA, - availability_zones=["AZ1", "az2"], - cpu_count=128, - gpu_count=8, - ) - backend = Mock(TYPE=BackendType.DSTACK) - backend.compute.return_value.get_offers.return_value = (offer for _ in range(1000)) - with ( - patch( - "dstack._internal.server.services.backends.get_project_backends", - return_value=[backend], - ), - patch.object(InstanceOfferWithAvailability, "model_copy", side_effect=AssertionError), - ): - result = await get_offers_by_requirements( - project=Mock(), - profile=Profile(availability_zones=["az1"]), - requirements=Requirements(resources=ResourcesSpec()), - blocks="auto", - exclude_not_available=True, - ) - assert result.offers == [] - assert result.unavailable_reason == InstanceAvailability.NO_QUOTA - assert offer.availability_zones == ["AZ1", "az2"] - assert offer.total_blocks == 1 - - @pytest.mark.asyncio - async def test_only_collects_unavailable_offers_in_the_volume_location(self): - backend = Mock(TYPE=BackendType.DSTACK) - backend.compute.return_value.get_offers.return_value = [ - get_instance_offer_with_availability(availability=InstanceAvailability.NO_QUOTA), - get_instance_offer_with_availability( - availability=InstanceAvailability.NOT_AVAILABLE, region="other-region" - ), - get_instance_offer_with_availability( - availability=InstanceAvailability.NOT_AVAILABLE, backend=BackendType.GCP - ), - ] - with patch( - "dstack._internal.server.services.backends.get_project_backends", - return_value=[backend], - ): - result = await get_offers_by_requirements( - project=Mock(), - profile=Profile(), - requirements=Requirements(resources=ResourcesSpec()), - exclude_not_available=True, - volumes=[[get_volume(configuration=get_volume_configuration(region="eu-west"))]], - ) - assert result.offers == [] - assert result.unavailable_reason == InstanceAvailability.NO_QUOTA - - @pytest.mark.asyncio - async def test_only_collects_unavailable_offers_suitable_for_the_placement_group(self): - backend = Mock(TYPE=BackendType.DSTACK) - compute = Mock(spec=ComputeMockSpec) - backend.compute.return_value = compute - compute.is_suitable_placement_group.side_effect = ( - lambda pg, offer: offer.region == pg.configuration.region - ) - compute.get_offers.return_value = [ - get_instance_offer_with_availability(availability=InstanceAvailability.NO_QUOTA), - get_instance_offer_with_availability( - availability=InstanceAvailability.NOT_AVAILABLE, region="other-region" - ), - ] - with patch( - "dstack._internal.server.services.backends.get_project_backends", - return_value=[backend], - ): - result = await get_offers_by_requirements( - project=Mock(), - profile=Profile(), - requirements=Requirements(resources=ResourcesSpec()), - exclude_not_available=True, - placement_group=PlacementGroup( - name="placement-group", - project_name="test_project", - configuration=get_placement_group_configuration(region="eu-west"), - ), - ) - assert result.offers == [] - assert result.unavailable_reason == InstanceAvailability.NO_QUOTA - - @pytest.mark.asyncio - @pytest.mark.parametrize("feature", ["multinode", "privileged", "instance_mounts"]) - async def test_only_collects_unavailable_offers_with_required_features(self, feature): - backend = Mock(TYPE=BackendType.DSTACK) - backend.compute.return_value.get_offers.return_value = [ - get_instance_offer_with_availability(availability=InstanceAvailability.NO_QUOTA), - get_instance_offer_with_availability( - backend=BackendType.DAYTONA, availability=InstanceAvailability.NOT_AVAILABLE - ), - ] - with patch( - "dstack._internal.server.services.backends.get_project_backends", - return_value=[backend], - ): - result = await get_offers_by_requirements( - project=Mock(), - profile=Profile(), - requirements=Requirements(resources=ResourcesSpec()), - exclude_not_available=True, - **{feature: True}, - ) - assert result.offers == [] - assert result.unavailable_reason == InstanceAvailability.NO_QUOTA - - @pytest.mark.asyncio - async def test_preserves_available_offer_order_when_a_backend_is_not_sorted_by_price(self): - vastai = Mock(TYPE=BackendType.VASTAI) - first = get_instance_offer_with_availability(backend=BackendType.VASTAI, price=1) - blocked = get_instance_offer_with_availability( - backend=BackendType.VASTAI, price=100, availability=InstanceAvailability.NO_QUOTA - ) - second = get_instance_offer_with_availability(backend=BackendType.VASTAI, price=2) - vastai.compute.return_value.get_offers.return_value = [first, blocked, second] - aws = Mock(TYPE=BackendType.AWS) - third = get_instance_offer_with_availability(price=3) - aws.compute.return_value.get_offers.return_value = [third] - with patch( - "dstack._internal.server.services.backends.get_project_backends", - return_value=[vastai, aws], - ): - result = await get_offers_by_requirements( - project=Mock(), - profile=Profile(), - requirements=Requirements(resources=ResourcesSpec()), - exclude_not_available=True, - ) - assert result.offers == [(vastai, first), (vastai, second), (aws, third)] - @pytest.mark.asyncio async def test_returns_all_offers(self): profile = Profile(name="test") @@ -313,10 +37,7 @@ async def test_returns_all_offers(self): requirements=requirements, ) m.assert_awaited_once() - assert res.offers == [ - (aws_backend_mock, aws_offer), - (runpod_backend_mock, runpod_offer), - ] + assert res == [(aws_backend_mock, aws_offer), (runpod_backend_mock, runpod_offer)] @pytest.mark.asyncio async def test_returns_multinode_offers(self): @@ -339,7 +60,7 @@ async def test_returns_multinode_offers(self): multinode=True, ) m.assert_awaited_once() - assert res.offers == [(aws_backend_mock, aws_offer)] + assert res == [(aws_backend_mock, aws_offer)] @pytest.mark.asyncio async def test_returns_volume_offers(self): @@ -378,7 +99,7 @@ async def test_returns_volume_offers(self): ], ) m.assert_awaited_once() - assert res.offers == [(runpod_backend_mock, runpod_offer2)] + assert res == [(runpod_backend_mock, runpod_offer2)] @pytest.mark.asyncio async def test_global_volume_matches_cpu_and_gpu_regions(self): @@ -403,7 +124,7 @@ async def test_global_volume_matches_cpu_and_gpu_regions(self): volumes=[[get_volume(configuration=DaytonaVolumeConfiguration())]], ) - assert offers.offers == [(daytona_backend, cpu_offer), (daytona_backend, gpu_offer)] + assert offers == [(daytona_backend, cpu_offer), (daytona_backend, gpu_offer)] @pytest.mark.asyncio async def test_returns_volume_offers_without_region(self): @@ -435,7 +156,7 @@ async def test_returns_volume_offers_without_region(self): volumes=[[get_volume(configuration=get_kubernetes_volume_configuration())]], ) m.assert_awaited_once() - assert res.offers == [(kubernetes_backend_mock, kubernetes_offer)] + assert res == [(kubernetes_backend_mock, kubernetes_offer)] @pytest.mark.asyncio async def test_returns_az_offers(self): @@ -471,10 +192,7 @@ async def test_returns_az_offers(self): requirements=requirements, ) m.assert_awaited_once() - assert res.offers == [ - (aws_backend_mock, aws_offer1), - (aws_backend_mock, expected_aws_offer3), - ] + assert res == [(aws_backend_mock, aws_offer1), (aws_backend_mock, expected_aws_offer3)] @pytest.mark.asyncio async def test_returns_az_offers_ignoring_case(self): @@ -496,7 +214,7 @@ async def test_returns_az_offers_ignoring_case(self): profile=profile, requirements=requirements, ) - assert res.offers == [(aws_backend_mock, expected_aws_offer)] + assert res == [(aws_backend_mock, expected_aws_offer)] @pytest.mark.asyncio async def test_returns_no_offers_for_multinode_instance_mounts_and_non_multinode_backend(self): @@ -521,4 +239,4 @@ async def test_returns_no_offers_for_multinode_instance_mounts_and_non_multinode instance_mounts=True, ) m.assert_awaited_once() - assert res.offers == [] + assert res == []