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 == []