o biw1@sddlZddlZddlZddlmZddlmZddlmZddl m Z m Z m Z m Z mZddlZddlmZmZmZmZddlmZddlmZmZmZmZdd lmZdd lmZdd lm Z e!e"Z#dddZ$Gddde Z%dS)Nwraps)RLock) ModuleType)AnyDictListOptionalTuple) bootstrap_gcp&construct_clients_from_provider_config get_node_typetpu_accelerator_config_to_type)GCPTPU) GCPComputeGCPNode GCPNodeType GCPResource)TPUCommandRunner)CommandRunnerInterface) NodeProvidercstfdd}|S)a!Retry decorator for methods of GCPNodeProvider. Upon catching BrokenPipeError, API clients are rebuilt and decorated methods are retried. Work-around for https://github.com/ray-project/ray/issues/16072. Based on https://github.com/kubeflow/pipelines/pull/5250/files. csxd}|kr:z |g|Ri|WSty3td|d7}|kr0|tnYnw|ksdSdS)Nrz#Caught a BrokenPipeError. Retrying.r)BrokenPipeErrorloggerwarning_construct_clientstimesleep)selfargskwargs try_count backoff_s max_triesmethod]/home/ubuntu/.local/lib/python3.10/site-packages/ray/autoscaler/_private/gcp/node_provider.pymethod_with_retries-s   z#_retry..method_with_retriesr)r&r%r$r)r'r#r(_retry#s r*cseZdZdedefddZddZdedefd d Ze d efd d Z defddZ defddZ defddZ e dedefddZdefddZdefddZe dedededeeeffddZdefd d!Ze defd"d#Ze dedefd$d%Zdedefd&d'Zed(d)Zed*eeefdeeeffd+d,Z -d5d.eded/eeefded0ed1ed2eeeefde ffd3d4 Z!Z"S)6GCPNodeProviderprovider_config cluster_namecCs6t|||t|_||dd|_i|_dS)Ncache_stopped_nodesF)r__init__rlockrgetr. cached_nodes)rr,r-r'r'r(r/@s  zGCPNodeProvider.__init__cCspt|j\}}}}i|_t||jd|jd|j|jtj<|dur6t||jd|jd|j|jtj<dSdS)N project_idavailability_zone) r r, resourcesrr-rCOMPUTErTPU)r_computetpur'r'r(rJs$  z"GCPNodeProvider._construct_clients node_namereturncCs|jt|S)zReturn the resource responsible for the node, based on node_name. This expects the name to be in format '[NAME]-[UUID]-[TYPE]', where [TYPE] is either 'compute' or 'tpu' (see ``GCPNodeType``). )r5r name_to_type)rr;r'r'r($_get_resource_depending_on_node_namedsz4GCPNodeProvider._get_resource_depending_on_node_name tag_filterscCsn|j*g}|jD] }||}||7}q dd|D|_dd|DWdS1s0wYdS)NcSsi|]}|d|qSnamer'.0ir'r'r( vsz8GCPNodeProvider.non_terminated_nodes..cSsg|]}|dqSr@r'rBr'r'r( wsz8GCPNodeProvider.non_terminated_nodes..)r0r5valueslist_instancesr2)rr? instancesresourcenode_instancesr'r'r(non_terminated_nodesls   $z$GCPNodeProvider.non_terminated_nodesnode_idcC<|j||}|WdS1swYdSN)r0_get_cached_node is_runningrrMnoder'r'r(rQy $zGCPNodeProvider.is_runningcCrNrO)r0rP is_terminatedrRr'r'r(rU~rTzGCPNodeProvider.is_terminatedcCrNrO)r0rP get_labelsrRr'r'r( node_tagsrTzGCPNodeProvider.node_tagstagscCsT|j|}||}||}|j||d}|WdS1s#wYdS)N)rSlabels)r0 _get_noder> set_labels)rrMrXrYrSrJresultr'r'r( set_node_tagss  $zGCPNodeProvider.set_node_tagscCZ|j ||}|}|dur||}|}|WdS1s&wYdSrO)r0rPget_external_iprZrrMrSipr'r'r( external_ip  $zGCPNodeProvider.external_ipcCr^rO)r0rPget_internal_iprZr`r'r'r( internal_iprczGCPNodeProvider.internal_ip base_configcountc s|jd|}t|}|j|i}|jr@|d|d|dd}|dd|}|r@fdd|D} || |t|8}|d kr_|||} d d| D} || Wd|SWd|S1sjwY|S) zCreates instances. Returns dict mapping instance id to each create operation result for the created instances. ray-node-name ray-node-typeray-user-node-type)rhrirjTNcs i|] }|d|dqSr@)start_instance)rCnrJr'r(rEsz/GCPNodeProvider.create_node..rcSsi|]\}}||qSr'r')rCr\ instance_idr'r'r(rEs )r0r r5r.rHupdatelencreate_instances) rrfrXrgrY node_type all_nodesfilters reuse_nodesreused_nodes_dictresultscreated_nodes_dictr'rmr( create_nodes@          zGCPNodeProvider.create_nodec Cstd|||}z |j|d}W|Stjjy@}z|jj dkr2t d|dd}n|dWYd}~|Sd}~ww)Nz"NodeProvider: {}: Terminating noderM!Tried to delete the node with id  but it was already gone.) rinfoformatr>delete_instancegoogleapiclienterrors HttpErrorrespstatusr)rrMrJr\ http_errorr'r'r(_thread_unsafe_terminate_nodes$     z-GCPNodeProvider._thread_unsafe_terminate_nodec Cs|jU||}z|jr ||}|r|j|d}n d}n|j|d}Wn'tjj yN}z|j j dkrAt d|dn|dWYd}~nd}~ww|WdS1s[wYdS)Nrzr{r|r})r0r>r.rPrQ stop_instancerrrrrrrr)rrMrJrSr\rr'r'r(terminate_nodes.     $zGCPNodeProvider.terminate_nodecCsp|i|j&||jvr|j|WdS||}|j|d}|WdS1s1wYdS)Nrz)rLr0r2r> get_instance)rrMrJinstancer'r'r(rZs    $zGCPNodeProvider._get_nodecCs||jvr |j|S||SrO)r2rZ)rrMr'r'r(rP s   z GCPNodeProvider._get_cached_nodecCst|SrO)r )cluster_configr'r'r(bootstrap_configsz GCPNodeProvider.bootstrap_configrcCsd|vr|St|}|d}|D]:}||d}t|tjkrKi}d}d|vr-|d}n d|vr7t|d}|s:qd|d|d<||d |q|S) aFill out TPU resources to the cluster config. To enable TPU pod autoscaling, we provide the TPU accelerator type as a resource that only exists on worker 0 of the pod slice. For instance, a v4-16 should have the resource labels: worker 0: resources = {"TPU": 4, "TPU-v4-16-head": 1} worker 1: resources = {"TPU": 4} For the autoscaler to correctly process the demands of creating a new TPU pod, then the autoscaler must know what a TPU pod is in the form of the TPU accelerator resource. Therefore we fill out TPU pods appropriately by providing the expected resource which we can deduce from the cluster config. available_node_types node_configacceleratorTypeacceleratorConfigrzTPU-z-headr5)copydeepcopyr rr7rro)rrrrrautodetected_resourcesaccelerator_typer'r'r(&fillout_available_node_types_resourcess.    z6GCPNodeProvider.fillout_available_node_types_resourcesN log_prefix auth_configprocess_runneruse_internal_ip docker_configc sj||}||} |||||||d} tj|jvr,||jtjkr,td| |d| Stjdi| S)z+Returns a TPU command runner as applicable.)rrrMrr-rr)rproviderNr')r>rrr7r5rsuperget_command_runner) rrrMrr-rrrrJr common_args __class__r'r(rBs  z"GCPNodeProvider.get_command_runnerrO)#__name__ __module__ __qualname__dictstrr/rrr>r*rLrQrUrWr]rbreintrryrrrrZrP staticmethodrrrrboolr rr __classcell__r'r'rr(r+?s^     $(     2  r+)rr)&rloggingr functoolsr threadingrtypesrtypingrrrr r r"ray.autoscaler._private.gcp.configr r r r ray.autoscaler._private.gcp.noderrrrr.ray.autoscaler._private.gcp.tpu_command_runnerrray.autoscaler.command_runnerrray.autoscaler.node_providerr getLoggerrrr*r+r'r'r'r(s"