Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion src/dstack/_internal/cli/services/configurators/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.")

Expand Down
13 changes: 12 additions & 1 deletion src/dstack/_internal/cli/utils/fleet.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,10 +8,15 @@
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
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

Expand Down Expand Up @@ -166,6 +171,12 @@ 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_MESSAGE, NO_BALANCE_MESSAGE)
):
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)
Expand Down
4 changes: 4 additions & 0 deletions src/dstack/_internal/core/consts.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
2 changes: 2 additions & 0 deletions src/dstack/_internal/core/models/instances.py
Original file line number Diff line number Diff line change
Expand Up @@ -246,6 +246,8 @@ class InstanceTerminationReason(str, Enum):
JOB_FINISHED = "job_finished"
UNREACHABLE = "unreachable"
NO_OFFERS = "no_offers"
# Recognize new reasons before storing them to support rolling upgrades.
NO_QUOTA = "no_quota"
MASTER_FAILED = "master_failed"
MAX_INSTANCES_LIMIT = "max_instances_limit"
FLEET_SPEC_MISMATCH = "fleet_spec_mismatch"
Expand Down
8 changes: 8 additions & 0 deletions src/dstack/_internal/core/models/runs.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
# 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"
INSTANCE_UNREACHABLE = "instance_unreachable"
INSTANCE_ACCESS_REVOKED = "instance_access_revoked"
Expand All @@ -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,
Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,8 @@
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 (
Expand All @@ -45,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,
Expand Down Expand Up @@ -236,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 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]:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -119,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 (
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -2569,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(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),
),
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,
Expand All @@ -2590,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


Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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),
)
)
14 changes: 13 additions & 1 deletion src/dstack/_internal/server/services/jobs/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -625,7 +629,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_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:
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:
Expand Down
Loading
Loading