o bi(@sddlZddlmZddlmZmZddlmZmZm Z m Z m Z ddl m Z mZmZddlmZmZmZmZmZddlmZddlmZmZmZmZmZeeZ eGd d d Z!eGd d d Z"Gd ddeZ#dS)N) defaultdict) dataclassfield)AnyDictListOptionalSet)DISABLE_LAUNCH_CONFIG_CHECK_KEYDISABLE_NODE_UPDATERS_KEYFOREGROUND_NODE_LAUNCH_KEY)NodeIDNodeIPNodeKind NodeStatusNodeType) NodeProvider)NODE_KIND_HEADTAG_RAY_NODE_KINDTAG_RAY_NODE_STATUSTAG_RAY_REPLICA_INDEXTAG_RAY_USER_NODE_TYPEc@sBeZdZUdZeedZeee fe d<ee dZ e ee d<dS) ScaleRequestzStores desired scale computed by the autoscaler. Attributes: desired_num_workers: Map of worker NodeType to desired number of workers of that type. workers_to_delete: List of ids of nodes that should be removed. )default_factorydesired_num_workersworkers_to_deleteN)__name__ __module__ __qualname____doc__rdictrrrint__annotations__setrr r r$r$Y/home/ubuntu/.local/lib/python3.10/site-packages/ray/autoscaler/batching_node_provider.pyrs rc@sFeZdZUdZeed<eed<eeed<e ed<dZ ee ed<dS)NodeDataaStores all data about a Ray node needed by the autoscaler. Attributes: kind: Whether the node is the head or a worker. type: The user-defined type of the node. replica_index: An identifier for nodes in a replica of a TPU worker group. This value is set as a Pod label by a GKE webhook when TPUs are requested ip: Cluster-internal ip of the node. ip can be None if the ip has not yet been assigned. status: The status of the node. You must adhere to the following semantics for status: * The status must be "up-to-date" if and only if the node is running. * The status must be "update-failed" if and only if the node is in an unknown or failed state. * If the node is in a pending (starting-up) state, the status should be a brief user-facing description of why the node is pending. kindtypeipstatusN replica_index) rrrrrr"rrrrr+strr$r$r$r%r&&s  r&c @seZdZdZdeeefdeddfddZdeee ffdd Z d e ddfd d Z d#d dZ deeefdeefddZddZdeeeffddZdedeeeffddZdedefddZdeeefdeeefdedeeeeffdd Zdedeeeeffd!d"ZdS)$BatchingNodeProvideraAbstract subclass of NodeProvider meant for use with external cluster managers. Batches reads of cluster state into a single method, get_node_data, called at the start of an autoscaling update. Batches modifications to cluster state into a single method, submit_scale_request, called at the end of an autoscaling update. Implementing a concrete subclass of BatchingNodeProvider only requires overriding get_node_data() and submit_scale_request(). See the method docstrings for more information. Note that an autoscaling update may be conditionally cancelled using the optional method safe_to_scale() of the root NodeProvider. provider_config cluster_namereturnNcCst|||i|_|tddusJdtd|tddus*Jdtd|tddus:Jdtdd|_t|_ t t t |_ dS)NFTz'To use BatchingNodeProvider, must set `z:True`.)r__init__node_data_dictgetr r r scale_change_neededr scale_requestrlistr,replica_index_to_nodes)selfr.r/r$r$r%r1Ts   zBatchingNodeProvider.__init__cCt)a8Queries cluster manager for node info. Returns a mapping from node id to NodeData. Each NodeData value must adhere to the semantics of the NodeData docstring. (Note in particular the requirements for NodeData.status.) Consistency requirement: If a node id was present in ScaleRequest.workers_to_delete of a previously submitted scale request, it should no longer be present as a key in get_node_data. (Node termination must be registered immediately when submit_scale_request returns.) NotImplementedErrorr8r$r$r% get_node_datatsz"BatchingNodeProvider.get_node_datar5cCr9)aTells the cluster manager which nodes to delete and how many nodes of each node type to maintain. Consistency requirement: If a node id was present in ScaleRequest.workers_to_delete of a previously submitted scale request, it should no longer be present as key in get_node_data. (Node termination must be registered immediately when submit_scale_request returns.) r:)r8r5r$r$r%submit_scale_requests z)BatchingNodeProvider.submit_scale_requestcCs|jr ||jd|_dS)z3Submit a scale request if it is necessary to do so.FN)r4r>r5r<r$r$r% post_processs  z!BatchingNodeProvider.post_process tag_filterscsd__ttd_tj}j |D]}j|j }|dur4j | |q fdd|D}|S)NF)rrcs&g|]}|kr|qSr$)items node_tags).0noder8r@r$r% s z=BatchingNodeProvider.non_terminated_nodes..) r4r=r2rcur_num_workersr#r5r6keysr7clearr+append)r8r@ all_nodesnode_idr+filtered_nodesr$rEr%non_terminated_nodess"    z)BatchingNodeProvider.non_terminated_nodescCs ||jS)zCReturns dict mapping node type to the number of nodes of that type.)_cur_num_workersr2r<r$r$r%rGs z$BatchingNodeProvider.cur_num_workersr2cCs8tt}|D]}|jtkrq||jd7<q|S)N)rr!valuesr'rr()r8r2num_workers_dict node_datar$r$r%rOs   z%BatchingNodeProvider._cur_num_workersrLcCs8|j|}t|jt|jt|ji}|jdur|j|t<|SN) r2rr'rr*rr(r+r)r8rLrStagsr$r$r%rBs   zBatchingNodeProvider.node_tagscCs |j|jSrT)r2r))r8rLr$r$r% internal_ips z BatchingNodeProvider.internal_ip node_configrUcountcCs&|t}|jj||7<d|_dS)NT)rr5rr4)r8rWrUrX node_typer$r$r% create_nodes z BatchingNodeProvider.create_nodecCs||jjvrtd|ddS||jvr!td|ddS|j|j}|jj|dkr7td|d|jj|d8<|jj|| |}t |vrv|t }|j |D]}||jjvru|jj|t d|d |d qZd |_ dS) Nz#Autoscaler tried to terminate node z8 twice in the same update. Skipping termination request.z*Autoscaler tried to terminate unkown node z. Skipping termination request.rz>NodeProvider attempted to request less than 0 workers of type rPzAutoscaler terminating node z in multi-host replica .T)r5rloggerwarningr2r(rAssertionErroraddrBrr7infor4)r8rLrYrUnode_replica_index worker_idr$r$r%terminate_nodes@        z#BatchingNodeProvider.terminate_node)r0N)rrrrrr,rr1r r&r=rr>r?rrNrGrOrBrVr!rrZrcr$r$r$r%r-As6         "r-)$logging collectionsr dataclassesrrtypingrrrrr !ray.autoscaler._private.constantsr r r ray.autoscaler._private.utilr rrrrray.autoscaler.node_providerrray.autoscaler.tagsrrrrr getLoggerrr\rr&r-r$r$r$r%s