&`i&ddlZddlmZmZddlmZmZddlmZddl m Z ddl m Z m Z mZmZmZmZmZmZejeZGddeZGd d ZdS) N)ABCabstractmethod)ListOptional) InstanceUtil)InstanceStorage)GetInstanceManagerStateReplyGetInstanceManagerStateRequestInstanceInstanceUpdateEventNodeKind StatusCodeUpdateInstanceManagerStateReply!UpdateInstanceManagerStateRequestc>eZdZdZedeeddfdZdS)InstanceUpdatedSubscriberz'Subscribers to instance status changes.eventsreturnNcdSN)selfrs /home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/autoscaler/v2/instance_manager/instance_manager.pynotifyz InstanceUpdatedSubscriber.notifys )__name__ __module__ __qualname____doc__rrr rrrrrrsL11 T"56 4   ^   rrc eZdZdZdedeeefdZde de fdZ de de fdZe dd ed ed ede fd ZededefdZededefdZedededefdZdS)InstanceManagera See `InstanceManagerService` in instance_manager.proto This handles updates to an instance, or inserts a new instance if it's an insert update. We should only be inserting new instances of the below statuses: 1. ALLOCATED: For unmanaged instance not initialized by InstanceManager, e.g. head node 2. QUEUED: For new instance being queued to launch. 3. TERMINATING: For leaked cloud instance that needs to be terminated. For full status transitions, see: https://docs.google.com/document/d/1NzQjA8Mh-oMc-QxXOa529oneWCoA8sDiVoNkBqqDb4U/edit#heading=h.k9a1sp4qpqj4 Not thread safe, should be used as a singleton. instance_storage"instance_status_update_subscribersc&||_|pg|_dSr)_instance_storage_status_update_subscribers)rr"r#s r__init__zInstanceManager.__init__/s "2*L*RPR'''rrequestrcd|jD}|j|\}}|jdkrS|j|krHd|jd|}t ||tj ||Sg}| D]P\}}||vr| |||} n| |} | | Q|j||} | js| j|krMd|d| j}t ||tj | j|Sd}t ||tj| j|S|jD]} | |j|tj| jS)ao Updates the instance manager state. If there's any failure, no updates would be made and the reply would contain the latest version of the instance manager state, and the error info. Args: request: The request to update the instance manager state. Returns: The reply to the request. ci|] }|j| Sr) instance_id).0updates r zAInstanceManager.update_instance_manager_state..IsSSS&,fSSSr) instance_idsrzVersion mismatch: expected: z , actual: )updatesexpected_storage_versionz"Failed to update instance storage.)r0r% get_instanceskeysexpected_versionloggerwarning_get_update_im_state_replyrVERSION_MISMATCHitems_update_instance_create_instanceappendbatch_upsert_instancessuccessversionerrorUNKNOWN_ERRORSr&rOK) rr(ids_to_updatesto_update_instancesr?err_strto_upsert_instancesr+r-instanceresult subscribers rupdate_instance_manager_statez-InstanceManager.update_instance_manager_state7s]$TS7?SSS'+'='K'K',,..(L( ( $W  #q ( (W-E-P-P%w/G%%"%%  NN7 # # #22+ !#1#7#7#9#9 1 1 K11100' 4f 0088  & &x 0 0 0 0'>>'%,?   ~ ~((V7VVfnVVw'''66/? W%%%66-v~w 9 / /J   go . . . ...z}fnMMMrct}|j\}}|jj|||j_tj |j _ |S)z Gets the instance manager state. Args: request: The request to get the instance manager state. Returns: The reply to the request. ) r r%r2state instancesextendvaluesr?rrBstatuscode)rr(replyrMr?s rget_instance_manager_statez*InstanceManager.get_instance_manager_statesh-..!3AACC 7 $$Y%5%5%7%7888% &M  r status_coder? error_messagecdt}||j_||_|r ||j_|S)a$ Returns a UpdateInstanceManagerStateReply with the given status code and version. Args: status_code: The status code. version: The version. error_message: The error message if any. Returns: The reply. )rrPrQr?message)rUr?rVrRs rr7z*InstanceManager._get_update_im_state_replys7 011'   1#0EL  rrGr-c|jtjkr|js Jd|jt jt jfvs Jd|js Jd|js Jd|j|_|j|_|j|_|j |_ dS|jtj kr|j s Jd|j |_ dS|jtj kr<|j s Jd|js Jd|j |_ |j|_dS|jtjkr|js JddSdS) z Apply status specific update to the instance. Args: instance: The instance to update. update: The update to apply. z,ALLOCATED update must have cloud_instance_idz6ALLOCATED update must have node_kind as WORKER or HEADz(ALLOCATED update must have instance_typez(RAY_RUNNING update must have ray_node_idz,REQUESTED update must have launch_request_idz(REQUESTED update must have instance_typez.TERMINATING update must have cloud instance idN)new_instance_statusr ALLOCATEDcloud_instance_id node_kindr WORKERHEAD instance_type ray_node_idnode_id RAY_RUNNING REQUESTEDlaunch_request_id TERMINATINGrGr-s r _apply_updatezInstanceManager._apply_updates  %); ; ;( > >= > >(# (H' S S)S S S'( > >= > >()/)AH &!'!1H %+%9H "%1H     '8+? ? ?% Q Q'Q Q Q%%1H     '8+= = =( > >= > >(' S S)S S S')/)AH &%+%9H " " "  '8+? ? ?( @ @? @ @(@ ? @ @rcx|js Jd|jtjtjtjfvs Jdt j|j|j |j|j }t t j ||t|||S)z> Create a new instance from the given update. z.upsert must be true for creating new instance.zPInvalid status for new instance, must be one of [ALLOCATED, QUEUED, TERMINATING])r+r`rPdetails)upsertrZr r[QUEUEDrfr new_instancer+r`rjr5infoget_log_str_for_updater!rh)r-rGs rr;z InstanceManager._create_instances }NNNNN})   O  .     /    ,* .-N      L7&IIJJJ%%h777rcrttj||tj||jsTJdt j|j dt j|jt |||S)z Update the instance with the given update. Args: instance: The instance to update. update: The update to apply. Returns: The updated instance. zInvalid status transition from z to ) r5rnrro set_statusrZr InstanceStatusNamerPr!rhrgs rr:z InstanceManager._update_instances  L7&IIJJJ&x1KLL   J&++HO<< J J&++F,FGG J J  L %%h777rN)rT)rrrrrrrrr'rrrJr r rS staticmethodrintstrr7r r rhr;r:rrrr!r!s"S)S-5T:S5T,USSSSIN8IN (ININININV5 %0DF*->A (\,%@%@2E%@%@%@\%@N!4\B85HX\rr!)loggingabcrrtypingrr)ray.autoscaler.v2.instance_manager.commonr3ray.autoscaler.v2.instance_manager.instance_storager'ray.core.generated.instance_manager_pb2r r r r r rrr getLoggerrr5rr!rrrr~s@########!!!!!!!!BBBBBBOOOOOO                      8 $ $        rrrrrrrrrrr