resources.py

"""Resources: compute requirements of Tasks."""
importdataclasses
importtextwrap
fromtypingimportAny, Dict, List, Optional, Set, Tuple, Union

importcolorama

fromskyimportcheckassky_check
fromskyimportclouds
fromskyimportexceptions
fromskyimportsky_logging
fromskyimportskypilot_config
fromsky.cloudsimportcloudassky_cloud
fromsky.cloudsimportservice_catalog
fromsky.provisionimportdocker_utils
fromsky.provision.kubernetesimportutilsaskubernetes_utils
fromsky.skyletimportconstants
fromsky.utilsimportaccelerator_registry
fromsky.utilsimportannotations
fromsky.utilsimportcommon_utils
fromsky.utilsimportconfig_utils
fromsky.utilsimportlog_utils
fromsky.utilsimportregistry
fromsky.utilsimportresources_utils
fromsky.utilsimportschemas
fromsky.utilsimportux_utils

logger=sky_logging.init_logger(__name__)

_DEFAULT_DISK_SIZE_GB=256

RESOURCE_CONFIG_ALIASES= {
'gpus': 'accelerators',
}


classResources:
"""Resources: compute requirements of Tasks.

 This class is immutable once created (to ensure some validations are done
 whenever properties change). To update the property of an instance of
 Resources, use ``resources.copy(**new_properties)``.

 Used:

 * for representing resource requests for tasks/apps
 * as a "filter" to get concrete launchable instances
 * for calculating billing
 * for provisioning on a cloud

 """
# If any fields changed, increment the version. For backward compatibility,
# modify the __setstate__ method to handle the old version.
_VERSION=22

def__init__(
self,
cloud: Optional[clouds.Cloud] =None,
instance_type: Optional[str] =None,
cpus: Union[None, int, float, str] =None,
memory: Union[None, int, float, str] =None,
accelerators: Union[None, str, Dict[str, int]] =None,
accelerator_args: Optional[Dict[str, str]] =None,
use_spot: Optional[bool] =None,
job_recovery: Optional[Union[Dict[str, Union[str, int]], str]] =None,
region: Optional[str] =None,
zone: Optional[str] =None,
image_id: Union[Dict[str, str], str, None] =None,
disk_size: Optional[int] =None,
disk_tier: Optional[Union[str, resources_utils.DiskTier]] =None,
ports: Optional[Union[int, str, List[str], Tuple[str]]] =None,
labels: Optional[Dict[str, str]] =None,
# Internal use only.
# pylint: disable=invalid-name
_docker_login_config: Optional[docker_utils.DockerLoginConfig] =None,
_docker_username_for_runpod: Optional[str] =None,
_is_image_managed: Optional[bool] =None,
_requires_fuse: Optional[bool] =None,
_cluster_config_overrides: Optional[Dict[str, Any]] =None,
 ):
"""Initialize a Resources object.

 All fields are optional. ``Resources.is_launchable`` decides whether
 the Resources is fully specified to launch an instance.

 Examples:
 .. code-block:: python

 # Fully specified cloud and instance type (is_launchable() is True).
 sky.Resources(clouds.AWS(), 'p3.2xlarge')
 sky.Resources(clouds.GCP(), 'n1-standard-16')
 sky.Resources(clouds.GCP(), 'n1-standard-8', 'V100')

 # Specifying required resources; the system decides the
 # cloud/instance type. The below are equivalent:
 sky.Resources(accelerators='V100')
 sky.Resources(accelerators='V100:1')
 sky.Resources(accelerators={'V100': 1})
 sky.Resources(cpus='2+', memory='16+', accelerators='V100')

 Args:
 cloud: the cloud to use.
 instance_type: the instance type to use.
 cpus: the number of CPUs required for the task.
 If a str, must be a string of the form ``'2'`` or ``'2+'``, where
 the ``+`` indicates that the task requires at least 2 CPUs.
 memory: the amount of memory in GiB required. If a
 str, must be a string of the form ``'16'`` or ``'16+'``, where
 the ``+`` indicates that the task requires at least 16 GB of memory.
 accelerators: the accelerators required. If a str, must be
 a string of the form ``'V100'`` or ``'V100:2'``, where the ``:2``
 indicates that the task requires 2 V100 GPUs. If a dict, must be a
 dict of the form ``{'V100': 2}`` or ``{'tpu-v2-8': 1}``.
 accelerator_args: accelerator-specific arguments. For example,
 ``{'tpu_vm': True, 'runtime_version': 'tpu-vm-base'}`` for TPUs.
 use_spot: whether to use spot instances. If None, defaults to
 False.
 job_recovery: the job recovery strategy to use for the managed
 job to recover the cluster from preemption. Refer to
 `recovery_strategy module <https://github.com/skypilot-org/skypilot/blob/master/sky/jobs/recovery_strategy.py>`__ # pylint: disable=line-too-long
 for more details.
 When a dict is provided, it can have the following fields:

 - strategy: the recovery strategy to use.
 - max_restarts_on_errors: the max number of restarts on user code
 errors.

 region: the region to use.
 zone: the zone to use.
 image_id: the image ID to use. If a str, must be a string
 of the image id from the cloud, such as AWS:
 ``'ami-1234567890abcdef0'``, GCP:
 ``'projects/my-project-id/global/images/my-image-name'``;
 Or, a image tag provided by SkyPilot, such as AWS:
 ``'skypilot:gpu-ubuntu-2004'``. If a dict, must be a dict mapping
 from region to image ID, such as:

 .. code-block:: python

 {
 'us-west1': 'ami-1234567890abcdef0',
 'us-east1': 'ami-1234567890abcdef0'
 }

 disk_size: the size of the OS disk in GiB.
 disk_tier: the disk performance tier to use. If None, defaults to
 ``'medium'``.
 ports: the ports to open on the instance.
 labels: the labels to apply to the instance. These are useful for
 assigning metadata that may be used by external tools.
 Implementation depends on the chosen cloud - On AWS, labels map to
 instance tags. On GCP, labels map to instance labels. On
 Kubernetes, labels map to pod labels. On other clouds, labels are
 not supported and will be ignored.
 _docker_login_config: the docker configuration to use. This includes
 the docker username, password, and registry server. If None, skip
 docker login.
 _docker_username_for_runpod: the login username for the docker
 containers. This is used by RunPod to set the ssh user for the
 docker containers.
 _requires_fuse: whether the task requires FUSE mounting support. This
 is used internally by certain cloud implementations to do additional
 setup for FUSE mounting. This flag also safeguards against using
 FUSE mounting on existing clusters that do not support it. If None,
 defaults to False.

 Raises:
 ValueError: if some attributes are invalid.
 exceptions.NoCloudAccessError: if no public cloud is enabled.
 """
self._version=self._VERSION
self._cloud=cloud
self._region: Optional[str] =region
self._zone: Optional[str] =zone

self._instance_type=instance_type

self._use_spot_specified=use_spotisnotNone
self._use_spot=use_spotifuse_spotisnotNoneelseFalse
self._job_recovery: Optional[Dict[str, Union[str, int]]] =None
ifjob_recoveryisnotNone:
ifisinstance(job_recovery, str):
job_recovery= {'strategy': job_recovery}
if'strategy'notinjob_recovery:
job_recovery['strategy'] =None

strategy_name=job_recovery['strategy']
ifstrategy_name=='none':
self._job_recovery=None
else:
ifstrategy_nameisnotNone:
job_recovery['strategy'] =strategy_name.upper()
self._job_recovery=job_recovery

ifdisk_sizeisnotNone:
ifround(disk_size) !=disk_size:
withux_utils.print_exception_no_traceback():
raiseValueError(
f'OS disk size must be an integer. Got: {disk_size}.')
self._disk_size=int(disk_size)
else:
self._disk_size=_DEFAULT_DISK_SIZE_GB

self._image_id=image_id
ifisinstance(image_id, str):
self._image_id= {self._region: image_id.strip()}
elifisinstance(image_id, dict):
ifNoneinimage_id:
self._image_id= {self._region: image_id[None].strip()}
else:
self._image_id= {
k.strip(): v.strip() fork, vinimage_id.items()
 }
self._is_image_managed=_is_image_managed

ifisinstance(disk_tier, str):
disk_tier_str=str(disk_tier).lower()
supported_tiers= [tier.valuefortierinresources_utils.DiskTier]
ifdisk_tier_strnotinsupported_tiers:
withux_utils.print_exception_no_traceback():
raiseValueError(f'Invalid disk_tier {disk_tier_str!r}. '
f'Disk tier must be one of '
f'{", ".join(supported_tiers)}.')
disk_tier=resources_utils.DiskTier(disk_tier_str)
self._disk_tier=disk_tier

ifportsisnotNone:
ifisinstance(ports, tuple):
ports=list(ports)
ifnotisinstance(ports, list):
ports= [ports]
ports=resources_utils.simplify_ports(
 [str(port) forportinports])
ifnotports:
# Set to None if empty. This is mainly for resources from
# cli, which will comes in as an empty tuple.
ports=None
self._ports=ports

self._labels=labels

self._docker_login_config=_docker_login_config

# TODO(andyl): This ctor param seems to be unused.
# We always use `Task.set_resources` and `Resources.copy` to set the
# `docker_username_for_runpod`. But to keep the consistency with
# `_docker_login_config`, we keep it here.
self._docker_username_for_runpod=_docker_username_for_runpod

self._requires_fuse=_requires_fuse

self._cluster_config_overrides=_cluster_config_overrides
self._cached_repr=None

self._set_cpus(cpus)
self._set_memory(memory)
self._set_accelerators(accelerators, accelerator_args)

defvalidate(self):
"""Validate the resources and infer the missing fields if possible."""
self._try_canonicalize_accelerators()
self._try_validate_and_set_region_zone()
self._try_validate_instance_type()
self._try_validate_cpus_mem()
self._try_validate_managed_job_attributes()
self._try_validate_image_id()
self._try_validate_disk_tier()
self._try_validate_ports()
self._try_validate_labels()

# When querying the accelerators inside this func (we call self.accelerators
# which is a @property), we will check the cloud's catalog, which can error
# if it fails to fetch some account specific catalog information (e.g., AWS
# zone mapping). It is fine to use the default catalog as this function is
# only for display purposes.
@service_catalog.fallback_to_default_catalog
def__repr__(self) ->str:
"""Returns a string representation for display.

 Examples:

 >>> sky.Resources(accelerators='V100')
 <Cloud>({'V100': 1})

 >>> sky.Resources(accelerators='V100', use_spot=True)
 <Cloud>([Spot], {'V100': 1})

 >>> sky.Resources(accelerators='V100',
 ... use_spot=True, instance_type='p3.2xlarge')
 AWS(p3.2xlarge[Spot], {'V100': 1})

 >>> sky.Resources(accelerators='V100', instance_type='p3.2xlarge')
 AWS(p3.2xlarge, {'V100': 1})

 >>> sky.Resources(instance_type='p3.2xlarge')
 AWS(p3.2xlarge, {'V100': 1})

 >>> sky.Resources(disk_size=100)
 <Cloud>(disk_size=100)
 """
ifself._cached_reprisnotNone:
returnself._cached_repr
accelerators=''
accelerator_args=''
ifself.acceleratorsisnotNone:
accelerators=f', {self.accelerators}'
ifself.accelerator_argsisnotNone:
accelerator_args=f', accelerator_args={self.accelerator_args}'

cpus=''
ifself._cpusisnotNone:
cpus=f', cpus={self._cpus}'

memory=''
ifself.memoryisnotNone:
memory=f', mem={self.memory}'

use_spot=''
ifself.use_spot:
use_spot='[Spot]'

image_id=''
ifself.image_idisnotNone:
ifNoneinself.image_id:
image_id=f', image_id={self.image_id[None]}'
else:
image_id=f', image_id={self.image_id}'

disk_tier=''
ifself.disk_tierisnotNone:
disk_tier=f', disk_tier={self.disk_tier.value}'

disk_size=''
ifself.disk_size!=_DEFAULT_DISK_SIZE_GB:
disk_size=f', disk_size={self.disk_size}'

ports=''
ifself.portsisnotNone:
ports=f', ports={self.ports}'

ifself._instance_typeisnotNone:
instance_type=f'{self._instance_type}'
else:
instance_type=''

# Do not show region/zone here as `sky status -a` would show them as
# separate columns. Also, Resources repr will be printed during
# failover, and the region may be dynamically determined.
hardware_str= (
f'{instance_type}{use_spot}'
f'{cpus}{memory}{accelerators}{accelerator_args}{image_id}'
f'{disk_tier}{disk_size}{ports}')
# It may have leading ',' (for example, instance_type not set) or empty
# spaces. Remove them.
whilehardware_strandhardware_str[0] in (',', ' '):
hardware_str=hardware_str[1:]

cloud_str='<Cloud>'
ifself.cloudisnotNone:
cloud_str=f'{self.cloud}'

self._cached_repr=f'{cloud_str}({hardware_str})'
returnself._cached_repr

@property
defrepr_with_region_zone(self) ->str:
region_str=''
ifself.regionisnotNone:
region_str=f', region={self.region}'
zone_str=''
ifself.zoneisnotNone:
zone_str=f', zone={self.zone}'
repr_str=str(self)
ifrepr_str.endswith(')'):
repr_str=repr_str[:-1] +f'{region_str}{zone_str})'
else:
repr_str+=f'{region_str}{zone_str}'
returnrepr_str

@property
defcloud(self):
returnself._cloud

@property
defregion(self):
returnself._region

@property
defzone(self):
returnself._zone

@property
definstance_type(self):
returnself._instance_type

@property
@annotations.lru_cache(scope='global', maxsize=1)
defcpus(self) ->Optional[str]:
"""Returns the number of vCPUs that each instance must have.

 For example, cpus='4' means each instance must have exactly 4 vCPUs,
 and cpus='4+' means each instance must have at least 4 vCPUs.

 (Developer note: The cpus field is only used to select the instance type
 at launch time. Thus, Resources in the backend's ResourceHandle will
 always have the cpus field set to None.)
 """
ifself._cpusisnotNone:
returnself._cpus
ifself.cloudisnotNoneandself._instance_typeisnotNone:
vcpus, _=self.cloud.get_vcpus_mem_from_instance_type(
self._instance_type)
returnstr(vcpus)
returnNone

@property
defmemory(self) ->Optional[str]:
"""Returns the memory that each instance must have in GB.

 For example, memory='16' means each instance must have exactly 16GB
 memory; memory='16+' means each instance must have at least 16GB
 memory.

 (Developer note: The memory field is only used to select the instance
 type at launch time. Thus, Resources in the backend's ResourceHandle
 will always have the memory field set to None.)
 """
returnself._memory

@property
@annotations.lru_cache(scope='global', maxsize=1)
defaccelerators(self) ->Optional[Dict[str, Union[int, float]]]:
"""Returns the accelerators field directly or by inferring.

 For example, Resources(AWS, 'p3.2xlarge') has its accelerators field
 set to None, but this function will infer {'V100': 1} from the instance
 type.
 """
ifself._acceleratorsisnotNone:
returnself._accelerators
ifself.cloudisnotNoneandself._instance_typeisnotNone:
returnself.cloud.get_accelerators_from_instance_type(
self._instance_type)
returnNone

@property
defaccelerator_args(self) ->Optional[Dict[str, str]]:
returnself._accelerator_args

@property
defuse_spot(self) ->bool:
returnself._use_spot

@property
defuse_spot_specified(self) ->bool:
returnself._use_spot_specified

@property
defjob_recovery(self) ->Optional[Dict[str, Union[str, int]]]:
returnself._job_recovery

@property
defdisk_size(self) ->int:
returnself._disk_size

@property
defimage_id(self) ->Optional[Dict[str, str]]:
returnself._image_id

@property
defdisk_tier(self) ->resources_utils.DiskTier:
returnself._disk_tier

@property
defports(self) ->Optional[List[str]]:
returnself._ports

@property
deflabels(self) ->Optional[Dict[str, str]]:
returnself._labels

@property
defis_image_managed(self) ->Optional[bool]:
returnself._is_image_managed

@property
defrequires_fuse(self) ->bool:
ifself._requires_fuseisNone:
returnFalse
returnself._requires_fuse

@property
defcluster_config_overrides(self) ->Dict[str, Any]:
ifself._cluster_config_overridesisNone:
return {}
returnself._cluster_config_overrides

@requires_fuse.setter
defrequires_fuse(self, value: Optional[bool]) ->None:
self._requires_fuse=value

@property
defdocker_username_for_runpod(self) ->Optional[str]:
returnself._docker_username_for_runpod

def_set_cpus(
self,
cpus: Union[None, int, float, str],
 ) ->None:
ifcpusisNone:
self._cpus=None
return

self._cpus=str(cpus)
ifisinstance(cpus, str):
ifcpus.endswith('+'):
num_cpus_str=cpus[:-1]
else:
num_cpus_str=cpus

try:
num_cpus=float(num_cpus_str)
exceptValueError:
withux_utils.print_exception_no_traceback():
raiseValueError(
f'The "cpus" field should be either a number or '
f'a string "<number>+". Found: {cpus!r}') fromNone
else:
num_cpus=float(cpus)

ifnum_cpus<=0:
withux_utils.print_exception_no_traceback():
raiseValueError(
f'The "cpus" field should be positive. Found: {cpus!r}')

def_set_memory(
self,
memory: Union[None, int, float, str],
 ) ->None:
ifmemoryisNone:
self._memory=None
return

self._memory=str(memory)
ifisinstance(memory, str):
ifmemory.endswith(('+', 'x')):
# 'x' is used internally for make sure our resources used by
# jobs controller (memory: 3x) to have enough memory based on
# the vCPUs.
num_memory_gb=memory[:-1]
else:
num_memory_gb=memory

try:
memory_gb=float(num_memory_gb)
exceptValueError:
withux_utils.print_exception_no_traceback():
raiseValueError(
f'The "memory" field should be either a number or '
f'a string "<number>+". Found: {memory!r}') fromNone
else:
memory_gb=float(memory)

ifmemory_gb<=0:
withux_utils.print_exception_no_traceback():
raiseValueError(
f'The "memory" field should be positive. Found: {memory!r}')

def_set_accelerators(
self,
accelerators: Union[None, str, Dict[str, int]],
accelerator_args: Optional[Dict[str, str]],
 ) ->None:
"""Sets accelerators.

 Args:
 accelerators: A string or a dict of accelerator types to counts.
 accelerator_args: A dict of accelerator types to args.
 """
ifacceleratorsisnotNone:
ifisinstance(accelerators, str): # Convert to Dict[str, int].
if':'notinaccelerators:
accelerators= {accelerators: 1}
else:
splits=accelerators.split(':')
parse_error= ('The "accelerators" field as a str '
'should be <name> or <name>:<cnt>. '
f'Found: {accelerators!r}')
iflen(splits) !=2:
withux_utils.print_exception_no_traceback():
raiseValueError(parse_error)
try:
num=float(splits[1])
num=int(num) ifnum.is_integer() elsenum
accelerators= {splits[0]: num}
exceptValueError:
withux_utils.print_exception_no_traceback():
raiseValueError(parse_error) fromNone

acc, _=list(accelerators.items())[0]
if'tpu'inacc.lower():
ifself.cloudisNone:
ifkubernetes_utils.is_tpu_on_gke(acc):
self._cloud=clouds.Kubernetes()
else:
self._cloud=clouds.GCP()
assert (self.cloud.is_same_cloud(clouds.GCP()) or
self.cloud.is_same_cloud(clouds.Kubernetes())), (
'Cloud must be GCP or Kubernetes for TPU '
'accelerators.')

ifaccelerator_argsisNone:
accelerator_args= {}

use_tpu_vm=accelerator_args.get('tpu_vm', True)
if (self.cloud.is_same_cloud(clouds.GCP()) and
notkubernetes_utils.is_tpu_on_gke(acc)):
if'runtime_version'notinaccelerator_args:

def_get_default_runtime_version() ->str:
ifnotuse_tpu_vm:
return'2.12.0'
# TPU V5 requires a newer runtime version.
ifacc.startswith('tpu-v5'):
return'v2-alpha-tpuv5'
# TPU V6e requires a newer runtime version.
elifacc.startswith('tpu-v6e'):
return'v2-alpha-tpuv6e'
return'tpu-vm-base'

accelerator_args['runtime_version'] = (
_get_default_runtime_version())
logger.info(
'Missing runtime_version in accelerator_args, using'
f' default ({accelerator_args["runtime_version"]})')

ifself.instance_typeisnotNoneanduse_tpu_vm:
ifself.instance_type!='TPU-VM':
withux_utils.print_exception_no_traceback():
raiseValueError(
'Cannot specify instance type (got '
f'{self.instance_type!r}) for TPU VM.')

self._accelerators=accelerators
self._accelerator_args=accelerator_args

defis_launchable(self) ->bool:
returnself.cloudisnotNoneandself._instance_typeisnotNone

defneed_cleanup_after_preemption_or_failure(self) ->bool:
"""Whether a resource needs cleanup after preemption or failure."""
assertself.is_launchable(), self
returnself.cloud.need_cleanup_after_preemption_or_failure(self)

def_try_canonicalize_accelerators(self) ->None:
"""Try to canonicalize the accelerators attribute.

 We don't canonicalize accelerators during creation of Resources object
 because it may check Kubernetes accelerators online. It requires
 Kubernetes credentias which may not be available locally when a remote
 API server is used.
 """
ifself._acceleratorsisNone:
return
self._accelerators= {
accelerator_registry.canonicalize_accelerator_name(
acc, self._cloud): acc_count
foracc, acc_countinself._accelerators.items()
 }

def_try_validate_and_set_region_zone(self) ->None:
"""Try to validate and set the region and zone attribute.

 Raises:
 ValueError: if the attributes are invalid.
 exceptions.NoCloudAccessError: if no public cloud is enabled.
 """
ifself._regionisNoneandself._zoneisNone:
return

ifself._cloudisNone:
# Try to infer the cloud from region/zone, if unique. If 0 or >1
# cloud corresponds to region/zone, errors out.
valid_clouds= []
enabled_clouds=sky_check.get_cached_enabled_clouds_or_refresh(
sky_cloud.CloudCapability.COMPUTE,
raise_if_no_cloud_access=True)
cloud_to_errors= {}
forcloudinenabled_clouds:
try:
cloud.validate_region_zone(self._region, self._zone)
exceptValueErrorase:
cloud_to_errors[repr(cloud)] =e
continue
valid_clouds.append(cloud)

ifnotvalid_clouds:
iflen(enabled_clouds) ==1:
cloud_str=f'for cloud {enabled_clouds[0]}'
else:
cloud_str=f'for any cloud among {enabled_clouds}'
withux_utils.print_exception_no_traceback():
iflen(cloud_to_errors) ==1:
# UX: if 1 cloud, don't print a table.
hint=list(cloud_to_errors.items())[0][-1]
else:
table=log_utils.create_table(['Cloud', 'Hint'])
table.add_row(['-----', '----'])
forcloud, errorincloud_to_errors.items():
reason_str='\n'.join(textwrap.wrap(
str(error), 80))
table.add_row([str(cloud), reason_str])
hint=table.get_string()
raiseValueError(
f'Invalid (region {self._region!r}, zone '
f'{self._zone!r}) {cloud_str}. Details:\n{hint}')
eliflen(valid_clouds) >1:
withux_utils.print_exception_no_traceback():
raiseValueError(
f'Cannot infer cloud from (region {self._region!r}, '
f'zone {self._zone!r}). Multiple enabled clouds '
f'have region/zone of the same names: {valid_clouds}. '
f'To fix: explicitly specify `cloud`.')
logger.debug(f'Cloud is not specified, using {valid_clouds[0]} '
f'inferred from region {self._region!r} and zone '
f'{self._zone!r}')
self._cloud=valid_clouds[0]

# Validate if region and zone exist in the catalog, and set the region
# if zone is specified.
self._region, self._zone=self._cloud.validate_region_zone(
self._region, self._zone)

defget_valid_regions_for_launchable(self) ->List[clouds.Region]:
"""Returns a set of `Region`s that can provision this Resources.

 Each `Region` has a list of `Zone`s that can provision this Resources.

 (Internal) This function respects any config in skypilot_config that
 may have restricted the regions to be considered (e.g., a
 ssh_proxy_command dict with region names as keys).
 """
assertself.is_launchable(), self

regions=self._cloud.regions_with_offering(self._instance_type,
self.accelerators,
self._use_spot,
self._region, self._zone)
ifself._image_idisnotNoneandNonenotinself._image_id:
regions= [rforrinregionsifr.nameinself._image_id]

# Filter the regions by the skypilot_config
ssh_proxy_command_config=skypilot_config.get_nested(
 (str(self._cloud).lower(), 'ssh_proxy_command'), None)
if (isinstance(ssh_proxy_command_config, str) or
ssh_proxy_command_configisNone):
# All regions are valid as the regions are not specified for the
# ssh_proxy_command config.
returnregions

# ssh_proxy_command_config: Dict[str, str], region_name -> command
# This type check is done by skypilot_config at config load time.
filtered_regions= []
forregioninregions:
region_name=region.name
ifregion_namenotinssh_proxy_command_config:
continue
# TODO: filter out the zones not available in the vpc_name
filtered_regions.append(region)

# Friendlier UX. Otherwise users only get a generic
# ResourcesUnavailableError message without mentioning
# ssh_proxy_command.
ifnotfiltered_regions:
yellow=colorama.Fore.YELLOW
reset=colorama.Style.RESET_ALL
logger.warning(
f'{yellow}Request {self} cannot be satisfied by any feasible '
'region. To fix, check that ssh_proxy_command\'s region keys '
f'include the regions to use.{reset}')

returnfiltered_regions

def_try_validate_instance_type(self) ->None:
"""Try to validate the instance type attribute.

 Raises:
 ValueError: if the attribute is invalid.
 exceptions.NoCloudAccessError: if no public cloud is enabled.
 """
ifself.instance_typeisNone:
return

# Validate instance type
ifself.cloudisnotNone:
valid=self.cloud.instance_type_exists(self._instance_type)
ifnotvalid:
withux_utils.print_exception_no_traceback():
raiseValueError(
f'Invalid instance type {self._instance_type!r} '
f'for cloud {self.cloud}.')
else:
# If cloud not specified
valid_clouds= []
enabled_clouds=sky_check.get_cached_enabled_clouds_or_refresh(
sky_cloud.CloudCapability.COMPUTE,
raise_if_no_cloud_access=True)
forcloudinenabled_clouds:
ifcloud.instance_type_exists(self._instance_type):
valid_clouds.append(cloud)
ifnotvalid_clouds:
iflen(enabled_clouds) ==1:
cloud_str=f'for cloud {enabled_clouds[0]}'
else:
cloud_str=f'for any cloud among {enabled_clouds}'
withux_utils.print_exception_no_traceback():
raiseValueError(
f'Invalid instance type {self._instance_type!r} '
f'{cloud_str}.')
iflen(valid_clouds) >1:
withux_utils.print_exception_no_traceback():
raiseValueError(
f'Ambiguous instance type {self._instance_type!r}. '
f'Please specify cloud explicitly among {valid_clouds}.'
 )
logger.debug(
f'Cloud is not specified, using {valid_clouds[0]} '
f'inferred from the instance_type {self.instance_type!r}.')
self._cloud=valid_clouds[0]

def_try_validate_cpus_mem(self) ->None:
"""Try to validate the cpus and memory attributes.

 Raises:
 ValueError: if the attributes are invalid.
 """
ifself._cpusisNoneandself._memoryisNone:
return
ifself._instance_typeisnotNone:
# The assertion should be true because we have already executed
# _try_validate_instance_type() before this method.
# The _try_validate_instance_type() method infers and sets
# self.cloud if self.instance_type is not None.
assertself.cloudisnotNone
cpus, mem=self.cloud.get_vcpus_mem_from_instance_type(
self._instance_type)
ifself._cpusisnotNone:
ifself._cpus.endswith('+'):
ifcpus<float(self._cpus[:-1]):
withux_utils.print_exception_no_traceback():
raiseValueError(
f'{self.instance_type} does not have enough '
f'vCPUs. {self.instance_type} has {cpus} '
f'vCPUs, but {self._cpus} is requested.')
elifcpus!=float(self._cpus):
withux_utils.print_exception_no_traceback():
raiseValueError(
f'{self.instance_type} does not have the requested '
f'number of vCPUs. {self.instance_type} has {cpus} '
f'vCPUs, but {self._cpus} is requested.')
ifself.memoryisnotNone:
ifself.memory.endswith(('+', 'x')):
ifmem<float(self.memory[:-1]):
withux_utils.print_exception_no_traceback():
raiseValueError(
f'{self.instance_type} does not have enough '
f'memory. {self.instance_type} has {mem} GB '
f'memory, but {self.memory} is requested.')
elifmem!=float(self.memory):
withux_utils.print_exception_no_traceback():
raiseValueError(
f'{self.instance_type} does not have the requested '
f'memory. {self.instance_type} has {mem} GB '
f'memory, but {self.memory} is requested.')

def_try_validate_managed_job_attributes(self) ->None:
"""Try to validate managed job related attributes.

 Raises:
 ValueError: if the attributes are invalid.
 """
ifself._job_recoveryisNoneorself._job_recovery['strategy'] isNone:
return
# Validate the job recovery strategy
registry.JOBS_RECOVERY_STRATEGY_REGISTRY.from_str(
self._job_recovery['strategy'])

defextract_docker_image(self) ->Optional[str]:
ifself.image_idisNone:
returnNone
iflen(self.image_id) ==1andself.regioninself.image_id:
image_id=self.image_id[self.region]
ifimage_id.startswith('docker:'):
returnimage_id[len('docker:'):]
returnNone

def_try_validate_image_id(self) ->None:
"""Try to validate the image_id attribute.

 Raises:
 ValueError: if the attribute is invalid.
 """
ifself._image_idisNone:
return

ifself.extract_docker_image() isnotNone:
# TODO(tian): validate the docker image exists / of reasonable size
ifself.cloudisnotNone:
self.cloud.check_features_are_supported(
self, {clouds.CloudImplementationFeatures.DOCKER_IMAGE})
return

ifself.cloudisNone:
withux_utils.print_exception_no_traceback():
raiseValueError(
'Cloud must be specified when image_id is provided.')

try:
self._cloud.check_features_are_supported(
self,
requested_features={
clouds.CloudImplementationFeatures.IMAGE_ID
 })
exceptexceptions.NotSupportedErrorase:
withux_utils.print_exception_no_traceback():
raiseValueError(
'image_id is only supported for AWS/GCP/Azure/IBM/OCI/'
'Kubernetes, please explicitly specify the cloud.') frome

ifself._regionisnotNone:
ifself._regionnotinself._image_id:
withux_utils.print_exception_no_traceback():
raiseValueError(
f'image_id {self._image_id} should contain the image '
f'for the specified region {self._region}.')
# Narrow down the image_id to the specified region.
self._image_id= {self._region: self._image_id[self._region]}

# Check the image_id's are valid.
forregion, image_idinself._image_id.items():
if (image_id.startswith('skypilot:') and
notself._cloud.is_image_tag_valid(image_id, region)):
region_str=f' ({region})'ifregionelse''
withux_utils.print_exception_no_traceback():
raiseValueError(
f'Image tag {image_id!r} is not valid, please make sure'
f' the tag exists in {self._cloud}{region_str}.')

if (self._cloud.is_same_cloud(clouds.AWS()) and
notimage_id.startswith('skypilot:') andregionisNone):
withux_utils.print_exception_no_traceback():
raiseValueError(
'image_id is only supported for AWS in a specific '
'region, please explicitly specify the region.')

# Validate the image exists and the size is smaller than the disk size.
forregion, image_idinself._image_id.items():
# Check the image exists and get the image size.
# It will raise ValueError if the image does not exist.
image_size=self.cloud.get_image_size(image_id, region)
ifimage_size>self.disk_size:
withux_utils.print_exception_no_traceback():
raiseValueError(
f'Image {image_id!r} is {image_size}GB, which is '
f'larger than the specified disk_size: {self.disk_size}'
' GB. Please specify a larger disk_size to use this '
'image.')

def_try_validate_disk_tier(self) ->None:
"""Try to validate the disk_tier attribute.

 Raises:
 ValueError: if the attribute is invalid.
 """
ifself.disk_tierisNone:
return
ifself.cloudisnotNone:
try:
self.cloud.check_disk_tier_enabled(self.instance_type,
self.disk_tier)
exceptexceptions.NotSupportedError:
withux_utils.print_exception_no_traceback():
raiseValueError(
f'Disk tier {self.disk_tier.value} is not supported '
f'for instance type {self.instance_type}.') fromNone

def_try_validate_ports(self) ->None:
"""Try to validate the ports attribute.

 Raises:
 ValueError: if the attribute is invalid.
 exceptions.NoCloudAccessError: if no public cloud is enabled.
 """
ifself.portsisNone:
return
ifself.cloudisnotNone:
self.cloud.check_features_are_supported(
self, {clouds.CloudImplementationFeatures.OPEN_PORTS})
else:
at_least_one_cloud_supports_ports=False