o ci@sddlZddlZddlZddlmZddlmZmZddlm Z m Z m Z m Z m Z mZddlmZmZeeZeGdddZGdd d Zed d Gd d d ZGdddZdS)N) defaultdict) dataclassfield)Callable DefaultDictDictHashableListOptional)*METRICS_PUSHER_GRACEFUL_SHUTDOWN_TIMEOUT_SSERVE_LOGGER_NAMEc@seZdZUeed<eed<dS) _MetricsTask task_func interval_sN)__name__ __module__ __qualname__r__annotations__floatrrT/home/ubuntu/.local/lib/python3.10/site-packages/ray/serve/_private/metrics_utils.pyr s  r c@seZdZdZdddeeegdffddZede j fdd Z d d Z d e fd dZd e dededdfddZddZddZdS) MetricsPusherz+Periodically runs registered asyncio tasks.N) async_sleeprcCs&|ptj|_t|_t|_d|_dSN)asynciosleep _async_sleepdict_tasks _async_tasks _stop_event)selfrrrr__init__s  zMetricsPusher.__init__returncCs|jdur t|_|jSr)r rEventr!rrr stop_event&s  zMetricsPusher.stop_eventcCs|jdSr)r&clearr%rrrstart-zMetricsPusher.startnamec st|j} |rdSz |j|Wnty6}zt d|d|WYd}~nd}~wwt| |j|j }tj||gtj dIdH|sX| q )zPeriodically runs `task_func` every `interval_s` until `stop_event` is set. If `task_func` raises an error, an exception will be logged. TNzFailed to run metrics task 'z': ) return_when)r create_taskr&waitdonerr Exceptionlogger exceptionrrFIRST_COMPLETEDcancel)r!r*wait_for_stop_evente sleep_taskrrr metrics_task0s*" zMetricsPusher.metrics_taskrrcCsFt|||j|<||jvs|j|r!t|||j|<dSdS)zRegister a task under the provided name, or update it. This method is idempotent - if a task is already registered with the specified name, it will update it with the most recent info. N)r rrr.rr,r7)r!r*rrrrrregister_or_update_taskKs z%MetricsPusher.register_or_update_taskcCs"|j|j|jdSr)r&setrr'rr%rrr stop_tasks[s  zMetricsPusher.stop_taskscsH|j|jrtjt|jtdIdH|j |j dS)zkShutdown metrics pusher gracefully. This method will ensure idempotency of shutdown call. )timeoutN) r&r9rrr-listvaluesr rr'r%rrrgraceful_shutdown`s   zMetricsPusher.graceful_shutdown)rrr__doc__r rintr"propertyrr$r&r(strr7r8r:r>rrrrrs*    rT)orderc@s(eZdZUeed<eddZeed<dS)TimeStampedValue timestampF)comparevalueN)rrrrrrrGrrrrrDqs rDc @seZdZdZddZdeeefdefddZdefd d Z d ed ed e efddZ dd ed ede d e efddZ dd ed ede fddZdS)InMemoryMetricsStorez-A very simple, in memory time series databasecCstt|_dSr)rr<datar%rrrr"zr)zInMemoryMetricsStore.__init__ data_pointsrEcCs0|D]\}}tj|j|t||dqdS)aNPush new data points to the store. Args: data_points: dictionary containing the metrics values. The key should uniquely identify this time series and to be used to perform aggregation. timestamp: the unix epoch timestamp the metrics are collected at. axN)itemsbisectinsortrIrD)r!rJrEr*rGrrradd_metrics_point}s z&InMemoryMetricsStore.add_metrics_pointstart_timestamp_scCsRt|jD]\}}t|dks|dj|kr|j|=q||||j|<qdS)a Prune keys and compact data that are outdated. For keys that haven't had new data recorded after the timestamp, remove them from the database. For keys that have, compact the datapoints that were recorded before the timestamp. rN)r<rIrNlenrE_get_datapoints)r!rRkey datapointsrrrprune_keys_and_compact_datas  z0InMemoryMetricsStore.prune_keys_and_compact_datarVwindow_start_timestamp_sr#cCs,|j|}tj|t|ddd}||dS)zz6InMemoryMetricsStore.window_average..)rUrIrTsumr!rVrYr[points_after_idxrrrwindow_averages   z#InMemoryMetricsStore.window_averagecCs0|||}|r ||j|<tdd|DddS)apPerform a max operation for metric `key`. Args: key: the metric name. window_start_timestamp_s: the unix epoch timestamp for the start of the window. The computed average will use all datapoints from this timestamp until now. do_compact: whether or not to delete the datapoints that's before `window_start_timestamp_s` to save memory. Default is true. Returns: Max value of the data points for the key on and after time window_start_timestamp_s, or None if there are no such points. csr\rr]r^rrrrarbz+InMemoryMetricsStore.max..N)default)rUrImaxrdrrrrhs  zInMemoryMetricsStore.maxN)T)rrrr?r"rrrrQrXr rUboolr rfrhrrrrrHws<  rH)rrOlogging collectionsr dataclassesrrtypingrrrrr r ray.serve._private.constantsr r getLoggerr0r rrDrHrrrrs   [