&`iB ddlZddlZddlZddlZddlZddlmZddlmZddl m Z ddl m Z m Z mZmZmZmZmZmZmZmZddlmZmZddlmZmZddlmZd ZejeZ eGd d Z!Gd d Z"GddZ# ddedee$dee$de$dee$f dZ% ddedede$dee$fdZ&deedefdZ'deeefdeeeffdZ(dS) N) defaultdict) dataclass)chain) AwaitableCallable DefaultDictDictHashableIterableListOptionalTupleUnion) TimeSeriesTimeStampedValue)*METRICS_PUSHER_GRACEFUL_SHUTDOWN_TIMEOUT_SSERVE_LOGGER_NAME)AggregationFunctionqueuedcDeZdZUeeegeffed<eed<dS) _MetricsTask task_func interval_sN)__name__ __module__ __qualname__rrr__annotations__floatt/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/serve/_private/metrics_utils.pyrr"s=XxI 667777r rceZdZdZdddeeegdffdZede j fdZ dZ d e fd Zd e d eeegeffd eddfd ZdZdZdS) MetricsPusherz+Periodically runs registered asyncio tasks.N) async_sleepr$c|p tj|_t|_t|_d|_dSN)asynciosleep _async_sleepdict_tasks _async_tasks _stop_event)selfr$s r!__init__zMetricsPusher.__init__+s: (87=/3vv 59VV59r returncN|jtj|_|jSr&)r-r'Eventr.s r! stop_eventzMetricsPusher.stop_event8s"   #&}D r c8|jdSr&)r4clearr3s r!startzMetricsPusher.start?s r namecKtj|j} |rdS |j|j}tj|r|d{Vn |n7#t$r*}t d|d|Yd}~nd}~wwxYwtj| |j|j }tj||gtj d{V|s|)zPeriodically runs `task_func` every `interval_s` until `stop_event` is set. If `task_func` raises an error, an exception will be logged. Supports both sync and async task functions. TNzFailed to run metrics task 'z': ) return_when)r' create_taskr4waitdoner+riscoroutinefunction Exceptionlogger exceptionr)rFIRST_COMPLETEDcancel)r.r8wait_for_stop_eventre sleep_tasks r! metrics_taskzMetricsPusher.metrics_taskBs&1$/2F2F2H2HII $"''))  N K-7 .y99 #)++%%%%%%%%IKKK N N N  !L!L!L!L!LMMMMMMMM N!,!!$+d"3">??J,01#3        ??$$ $!!###/ $sAB B< B77B<rrct|||j|<||jvs|j|r1t j|||j|<dSdS)aRegister a sync or async 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. Args: name: Unique name for the task. task_func: Either a sync function or async function (coroutine function). interval_s: Interval in seconds between task executions. N)rr+r,r=r'r;rG)r.r8rrs r!register_or_update_taskz%MetricsPusher.register_or_update_taskcsu")J?? D t( ( (D,=d,C,H,H,J,J (&-&9$:K:KD:Q:Q&R&RD d # # # ) (r c|j|j|jdSr&)r4setr+r6r,r3s r! stop_taskszMetricsPusher.stop_tasksxsE   !!!!!r c8K|j|jrEtjt |jtd{V|j |j dS)zkShutdown metrics pusher gracefully. This method will ensure idempotency of shutdown call. )timeoutN) r4rKr,r'r<listvaluesrr+r6r3s r!graceful_shutdownzMetricsPusher.graceful_shutdown}s    ,T&--//00B         !!!!!r )rrr__doc__r rintr/propertyr'r2r4r7strrGrrrIrLrQrr r!r#r#(s55 8< 9 9 9hud{34 9 9 9 9 GM   X    $s$$$$BSS8B M#::;S S  SSSS*""" """""r r#cNeZdZdZdZdeeefdefdZdefdZ ded ed e fd Z d e ed e e egefd eeeeffdZded eefdZd e ed eeeeffdZd e ed eeeeffdZded efdZdS)InMemoryMetricsStorez-A very simple, in memory time series databasec8tt|_dSr&)rrOdatar3s r!r/zInMemoryMetricsStore.__init__s7B47H7H r data_points timestampc|D]4\}}tj|j|t ||5dS)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)itemsbisectinsortrYr)r.rZr[r8values r!add_metrics_pointz&InMemoryMetricsStore.add_metrics_pointsZ',,.. S SKD% MDIdO/? 5/Q/Q R R R R R S Sr start_timestamp_sct|jD]P\}}t|dks|dj|kr |j|=2||||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)rOrYr`lenr[_get_datapoints)r.rekey datapointss r!prune_keys_and_compact_dataz0InMemoryMetricsStore.prune_keys_and_compact_datas $DIOO$5$566 N NOC:!##z"~'?BS'S'SIcNN!%!5!5c;L!M!M #  N Nr rjwindow_start_timestamp_sr0cz|j|}tj|t|d}||dS)z>> store = InMemoryMetricsStore() >>> store.data.update({ ... "a": [TimeStampedValue(0, 1.0), TimeStampedValue(1, 2.0)], ... "b": [], ... "c": [TimeStampedValue(0, 10.0)], ... }) Using sum across keys: >>> store._aggregate_reduce(keys=["a", "b", "c"], aggregate_fn=sum) (13.0, 2) Here: - The aggregated value is 1.0 + 2.0 + 10.0 = 13.0 - Only keys "a" and "c" contribute values, so report_count = 2 rc3xKD]3}j|g}|s dz |D] }|jV 4dS)zPGenerator that yields values from valid keys without storing them all in memory.NrYgetrc)rjseriestimestamp_valuerpr.valid_key_counts r!_values_generatorzAInMemoryMetricsStore._aggregate_reduce.._values_generatorsr 0 0sB//1$'-00O)/////0  0 0r )Nr)next StopIterationr)r.rprqrz values_gen first_valueaggregated_resultrys`` @r!_aggregate_reducez&InMemoryMetricsStore._aggregate_reducesD 0 0 0 0 0 0 0'&((  z**KK   77  )L }j)I)IJJ /11s ( 66cl|j|dsdS|j|djS)z%Get the latest value for a given key.Nrgru)r.rjs r! get_latestzInMemoryMetricsStore.get_latests5 y}}S$'' 4y~b!''r c8||tS)aSum the entire set of timeseries values across the specified keys. Args: keys: Iterable of keys to aggregate across. Returns: A tuple of (float, int) where the first element is the sum across all values found at `keys`, and the second is the number of valid keys used to compute the sum. Returns (None, 0) if no valid keys have data. )rsumr.rps r! aggregate_sumz"InMemoryMetricsStore.aggregate_sums%%dC000r cB||tjS)aAverage the entire set of timeseries values across the specified keys. Args: keys: Iterable of keys to aggregate across. Returns: A tuple of (float, int) where the first element is the mean across all values found at `keys`, and the second is the number of valid keys used to compute the mean. Returns (None, 0) if no valid keys have data. )r statisticsmeanrs r! aggregate_avgz"InMemoryMetricsStore.aggregate_avgs%%dJO<< \!!tNML## ?l * *!KM  ?j ( ( E%/:66 - a<< MH$< < h &N " j   ,  88("4BQ4F4F . 0 0DPr timeseriesaggregation_functionc|tjkrt||S|tjkr|rt d|DndS|tjkr|rt d|DndStd|)z@Aggregate the values in a timeseries using a specified function.)rc3$K|] }|jV dSr&rc.0tss r! z'aggregate_timeseries..x$1128111111r Nc3$K|] }|jV dSr&rrs r!rz'aggregate_timeseries..zrr zInvalid aggregation function: )rMEANrMAXmaxMINr ValueError)rrrs r!aggregate_timeseriesros 2777$Z}MMMM !4!8 8 85?Is11j111111TI !4!8 8 85?Is11j111111TIP:NPPQQQr replicas_timeseriescd|D}|sgSt|dkr|dSg}dgt|z}t|D]Z\}}|rSt|} t|}t j||j||j|fJ#t$rYVwxYw[g}d} |rt j |\} }} }||} | ||<| | | z z } t|} t j|| j|| j|fn#t$rYnwxYw| | krZt| d}|r%|dj|krt|| |d<n#| t|| ||S)aJ Merge multiple gauge time series (right-continuous, LOCF) into an instantaneous total time series as a step function. This approach treats each replica's gauge as right-continuous, last-observation- carried-forward (LOCF), which matches gauge semantics. It produces an exact instantaneous total across replicas without bias from arbitrary windowing. Uses a k-way merge algorithm for O(n log k) complexity where k is the number of timeseries and n is the total number of events. Timestamps are rounded to 10ms precision (2 decimal places) and datapoints with the same rounded timestamp are combined, keeping the most recent value. Args: replicas_timeseries: List of time series, one per replica. Each time series is a list of TimeStampedValue objects sorted by timestamp. Returns: A list of TimeStampedValue representing the instantaneous total at event times. Between events, the total remains constant (step function). Timestamps are rounded to 10ms precision and duplicate timestamps are combined. cg|]}||Srr)rrws r! z-merge_instantaneous_total..sHHHHVHHHr rtrrrg) rh enumerateiterr{heapqheappushr[rcr|heappoproundrappend)r active_series merge_heapcurrent_values replica_idxrwiterator first_pointmerged running_totalr[rc old_value next_pointrounded_timestamps r!merge_instantaneous_totalrs(6IH*=HHHM   =QQJUS///N )77   V  F||H "8nn  *K9JHU!     FM R27- 2K2K/ ;x";/ &+{#**  +/>>J N%{J4DhO        D  I   %i 3 3  R&*.2CCC-.?OOr  ./@-PPQQQ= R@ Ms$'2B B'&B' 2D D D timeseries_dictsctt}|D]7}|D] \}}|||!8d|DS)zU Merge multiple time-series dictionaries using instantaneous merge approach. c4i|]\}}|t|Sr)r)rrjts_lists r! z*merge_timeseries_dicts..s' W W W WC*733 W W Wr )rrOr`r)rrts_dictrjrs r!merge_timeseries_dictsrsz 1rs ######!!!!!!                        CBBBBBBB100000  , - -   c"c"c"c"c"c"c"c"L^^^^^^^^F%)"& =Q=Q=Q5/=Q=Q =Q e_ =Q=Q=Q=QF R R R- R Re_ R R R R Wj)WWWWWt X"8Z#78 X:%& X X X X X Xr