o ciI@@sddlZddlZddlmZddlmZmZmZmZddl m Z m Z m Z m Z ddlmZmZddlmZddlmZeeZeGdd d ZeGd d d ZGd d d ZGdddZdS)N) dataclass)DictListOptionalSet)DeploymentHandleSource DeploymentID ReplicaIDTargetCapacityDirection)&RAY_SERVE_MIN_HANDLE_METRICS_TIMEOUT_SSERVE_LOGGER_NAME)DeploymentInfo)"get_capacity_adjusted_num_replicasc@sjeZdZUdZeeed<eed<eed<e e efed<eed<e defdd Z e de fd d Zd S) HandleMetricReportaReport from a deployment handle on queued and ongoing requests. Args: actor_id: If the deployment handle (from which this metric was sent) lives on an actor, the actor ID of that actor. handle_source: Describes what kind of entity holds this deployment handle: a Serve proxy, a Serve replica, or unknown. queued_requests: The current number of queued requests at the handle, i.e. requests that haven't been assigned to any replica yet. running_requests: A map of replica ID to the average number of requests, assigned through the handle, running at that replica. timestamp: The time at which this report was received. actor_id handle_sourcequeued_requestsrunning_requests timestampreturncCs|jt|jS)z,Total number of queued and running requests.)rsumrvaluesselfrX/home/ubuntu/.local/lib/python3.10/site-packages/ray/serve/_private/autoscaling_state.pytotal_requests/sz!HandleMetricReport.total_requestscCs|jtjtjfvS)aQWhether the handle source is a Serve actor. More specifically, this returns whether a Serve actor tracked by the controller holds the deployment handle that sent this report. If the deployment handle lives on a driver, a Ray task, or an actor that's not a Serve replica, then this returns False. )rrPROXYREPLICArrrris_serve_component_source4s z,HandleMetricReport.is_serve_component_sourceN)__name__ __module__ __qualname____doc__rstr__annotations__rfloatrr propertyrboolrrrrrrs  rc@s"eZdZUdZeed<eed<dS)ReplicaMetricReportzReport from a replica on ongoing requests. Args: running_requests: Average number of running requests at the replica. timestamp: The time at which this report was received. rrN)r r!r"r#r&r%rrrrr)Cs  r)c@seZdZdZdefddZdededefdd Zd e fd d Z defd dZ defddZ de e fddZdefddZdedefddZd e deededdfddZded eed!ed"ed#ee efdeddfd$d%Zd&eeddfd'd(Z )d/ded*edefd+d,Zdefd-d.ZdS)0AutoscalingStatez,Manages autoscaling for a single deployment. deployment_idcCs>||_t|_t|_d|_d|_d|_g|_d|_d|_ dSN) _deployment_iddict_handle_requests_replica_requests_deployment_info_config_policy_running_replicas_target_capacity_target_capacity_directionrr+rrr__init__Ts zAutoscalingState.__init__infocurr_target_num_replicasrcCsl|jj}|jdus|j|r|jdur|j}n|}||_||_|j|_|j|_ |j |_ i|_ | |S)zyRegisters an autoscaling deployment's info. Returns the number of replicas the target should be set to. N)deployment_configautoscaling_configr1config_changedinitial_replicasr2 get_policyr3target_capacityr5target_capacity_directionr6 _policy_state apply_bounds)rr9r:configtarget_num_replicasrrrregisterfs   zAutoscalingState.register replica_idcCs||jvr |j|=dSdSr,)r0)rrGrrron_replica_stopped}s  z#AutoscalingState.on_replica_stoppedcCs8|jjdur|jtjkrt|jj|jSt|jj|jSr,)r2r>r6r UPrr5 min_replicasrrrrget_num_replicas_lower_bounds  z-AutoscalingState.get_num_replicas_lower_boundcCst|jj|jSr,)rr2 max_replicasr5rrrrget_num_replicas_upper_boundsz-AutoscalingState.get_num_replicas_upper_boundrunning_replicascCs ||_dS)z=Update cached set of running replica IDs for this deployment.N)r4)rrNrrrupdate_running_replica_idss z+AutoscalingState.update_running_replica_ids&num_replicas_running_at_target_versioncCs|||kS)zWhether or not this deployment is within the autoscaling bounds. Returns: True if the number of running replicas for the current deployment version is within the autoscaling bounds. False otherwise. )rC)rrPrrris_within_boundss z!AutoscalingState.is_within_bounds num_replicascCst|t||S)zqClips a replica count with current autoscaling bounds. This takes into account target capacity. )maxrKminrM)rrRrrrrCs zAutoscalingState.apply_bounds window_avgsend_timestampNcCs@|durdS||jvs||j|jkrt||d|j|<dSdS)z8Records average number of ongoing requests at a replica.N)rr)r0rr))rrGrUrVrrr"record_request_metrics_for_replicas z3AutoscalingState.record_request_metrics_for_replica handle_idrrrrcCs:||jvs ||j|jkrt|||||d|j|<dSdS)zgRecords average number of queued and running requests at a handle for this deployment. )rrrrrN)r/rr)rrXrrrrrVrrr!record_request_metrics_for_handles z2AutoscalingState.record_request_metrics_for_handlealive_serve_actor_idsc Cstd|jjt}t|jD]_\}}|jr<|jdur<|j|vr<|j|=|j dkr;t d|d|jd|j dqt |j |kro|j|=|j dkro|j}|rYd|d nd }t d |d |d |d d|j d qdS)Drops handle metrics that are no longer valid. This includes handles that live on Serve Proxy or replica actors that have died AND handles from which the controller hasn't received an update for too long. NrzDropping metrics for handle 'z%' because the Serve actor it was on (z) is no longer alive. It had z ongoing requestsz on actor 'z' z#Dropping stale metrics for handle 'z#because no update was received for z.1fzs. Ongoing requests was: .)rSr2metrics_interval_sr listr/itemsrrrloggerdebugtimerr9)rrZ timeout_srX handle_metricr actor_inforrrdrop_stale_handle_metricssB     z*AutoscalingState.drop_stale_handle_metricsF_skip_bound_checkc Cs@|j||t|j|j|||jd}|r|S||S)aDecide the target number of replicas to autoscale to. The decision is based off of the number of requests received for this deployment. After the decision number of replicas is returned by the policy, it is then bounded by the bounds min and max adjusted by the target capacity and returned. If `_skip_bound_check` is True, then the bounds are not applied. )r:total_num_requestsnum_running_replicasrDcapacity_adjusted_min_replicascapacity_adjusted_max_replicas policy_state) r3get_total_num_requestslenr4r2rKrMrBrC)rr:ridecision_num_replicasrrrget_decision_num_replicass   z*AutoscalingState.get_decision_num_replicascCsxd}|jD]}||jvr||j|j7}q|dk}|jD]}||j7}|s9|jD]}||jvr8||j|7}q*q|S)aGet average total number of requests aggregated over the past `look_back_period_s` number of seconds. If there are 0 running replicas, then returns the total number of requests queued at handles This code assumes that the metrics are either emmited on handles or on replicas, but not both. Its the responsibility of the writer to ensure enclusivity of the metrics. r)r4r0rr/rr)rridmetrics_collected_on_replicasrfrrrro s     z'AutoscalingState.get_total_num_requests)F)r r!r"r#rr8r intrFr rHrKrMrrOrQrCrr&rWr$rrrYrrhr(rrrorrrrr*QsX     * r*c@s"eZdZdZddZdedededefdd Zdefd d Z ded e e fd dZ de fddZ deeeffddZdededefddZdedefddZdededefddZde deededdfddZded ed!eed"ed#ed$ee efdeddfd%d&Zd'eeddfd(d)ZdS)*AutoscalingStateManagerzManages all things autoscaling related. Keeps track of request metrics for each deployment and decides on the target number of replicas to autoscale to based on those metrics. cCs i|_dSr,_autoscaling_statesrrrrr8Es z AutoscalingStateManager.__init__r+r9r:rcCs6|jjsJ||jvrt||j|<|j|||S)z%Register autoscaling deployment info.)r;r<rxr*rF)rr+r9r:rrrregister_deploymentHs   z+AutoscalingStateManager.register_deploymentcCs|j|ddS)z Remove deployment from tracking.N)rxpopr7rrrderegister_deploymentVsz-AutoscalingStateManager.deregister_deploymentrNcCs|j||dSr,)rxrO)rr+rNrrrrOZs z2AutoscalingStateManager.update_running_replica_idsrGcCs(|j}||jvr|j||dSdSr,)r+rxrH)rrGr+rrrrHas z*AutoscalingStateManager.on_replica_stoppedcsfddjDS)Ncsi|]}||qSr)ro).0r+rrr gs z7AutoscalingStateManager.get_metrics..rwrrrr get_metricsfs z#AutoscalingStateManager.get_metricscCs|j|j|dS)N)r:)rxrr)rr+r:rrrget_target_num_replicasls z/AutoscalingStateManager.get_target_num_replicascCs|j|Sr,)rxror7rrrrossz.AutoscalingStateManager.get_total_num_requestsrPcCs|j||Sr,)rxrQ)rr+rPrrrrQvs z(AutoscalingStateManager.is_within_boundsrUrVNcCs.|j}||jvr|j|j|||ddSdS)N)rGrUrV)r+rxrW)rrGrUrVr+rrrrW}s   z:AutoscalingStateManager.record_request_metrics_for_replicarXrrrrcCs.||jvr|j|j||||||ddSdS)z,Update request metric for a specific handle.)rXrrrrrVN)rxrY)rr+rXrrrrrVrrrrYs  z9AutoscalingStateManager.record_request_metrics_for_handlerZcCs|jD]}||qdS)r[N)rxrrh)rrZautoscaling_staterrrrhs z1AutoscalingStateManager.drop_stale_handle_metrics)r r!r"r#r8rr ruryr{rr rOrHrr&r~rror(rQrrWr$rrYrrhrrrrrv>s|         rv)loggingrd dataclassesrtypingrrrrray.serve._private.commonrrr r ray.serve._private.constantsr r "ray.serve._private.deployment_infor ray.serve._private.utilsr getLoggerrbrr)r*rvrrrrs     , n