yjUdZddlmZddlZddlZddlZddlZddlmZm Z m Z ej e Z dZdZGddZdad ed <ejZdd ZddZdddZeZgdZdS)uMonitoring emitter: fire-and-forget queue + background dispatcher. The emitter is the single seam between producers (gateway status hooks, the diagnostic log handler) and consumers (the OTLP streamers). Its contract is the hot-path invariant: ``emit()`` MUST return in O(microseconds), MUST NOT block on disk/network, and MUST NEVER raise into the caller. A monitoring failure is logged locally and dropped — it can never affect the gateway or a session. Mechanism: * ``emit(event)`` does a non-blocking ``queue.put_nowait`` wrapped in a bare except. On a full queue it drops the *oldest* event and counts the drop. * A daemon thread drains the queue and fans each batch out to subscribers (the OTLP metric/span/log streamers). Each subscriber is fail-isolated — a slow or raising subscriber never affects the hot path or its peers. Nothing is persisted here. Monitoring is an egress path, not a local store; if no subscriber is attached, events simply age out of the ring buffer. ) annotationsN)AnyDictOptionali'cjeZdZdZddddZdd Zdd Zdd ZddZddZ ddZ dddZ ddZ ddZ dS)MonitoringEmitterz?Owns the queue, the dispatcher thread, and the subscriber list.Tenabledr boolreturnNonec||_tjt|_d|_d|_tj|_ d|_ tj |_ d|_ g|_dS)N)maxsizerF)_enabledqueueQueue _MAX_QUEUE_q_dropped _dispatched threadingEvent_stop_startedLock_lock_thread _subscribers)selfr s >/home/ragecks/.hermes/hermes-agent/agent/monitoring/emitter.py__init__zMonitoringEmitter.__init__'sh 16Z1P1P1P _&&  ^%% 37 #%eventrc|jsdS t|dr|nt|}|dt j| |j |dS#tj $r |j |j |xjdz c_|j |n #t$r|xjdz c_YnwxYwYdSYdSwxYw#t$r t ddYdSwxYw)zEnqueue an event. Never blocks, never raises. ``event`` may be a dataclass with ``to_dict()`` or a plain dict. Nto_dictts_nszmonitoring emit failedTexc_info)rhasattrr&dict setdefaulttimetime_ns_ensure_startedr put_nowaitrFull get_nowait task_doner Exceptionloggerdebug)r r$payloads r!emitzMonitoringEmitter.emit5s }  F B)0 )B)BSemmoooU G   w  7 7 7  " " " '""7+++++: ' ' ''G&&(((G%%'''MMQ&MMG&&w//// '''MMQ&MMMM'0//!MM ' B B B LL1DL A A A A A A Bs[A.D-:BD*&ADD*D D*D  D*#D-&D-)D**D--&EEc|jrdS|j5|jr ddddStj|jdd|_|jd|_ddddS#1swxYwYdS)Nzhermes-monitoring-dispatchTtargetnamedaemon)rrrThread_runrstartr s r!r0z!MonitoringEmitter._ensure_startedOs =  F Z ! !}  ! ! ! ! ! ! ! !%+y'CDDL L    DM ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! !s A5AA55A9<A9cr|js |jd}n#tj$rYHwxYw|g}t |tkr[ ||j n#tj$rYnwxYwt |tk[ | ||D]}|j n##|D]}|j wxYw|jdSdS)Ng?timeout) ris_setrgetrEmptylen _DRAIN_BATCHappendr3 _dispatchr4)r firstbatch_s r!r@zMonitoringEmitter._run[sg*##%% (  C 00;    GEe**|++LL!3!3!5!56666{Ee**|++  (u%%%((AG%%''''(((AG%%''''(*##%% ( ( ( ( (s-8A  A ),BB('B(C88 Dct|jD]:} ||#t$rtddY7wxYw|xjt |z c_dS)Nzmonitoring subscriber failedTr))listrr5r6r7rrI)r rNsubs r!rLzMonitoringEmitter._dispatchms)** L LC LE  L L L ;d KKKKK L CJJ&s $&A  A cZ||jvr|j|d|_dS)z?Register a live batch subscriber (callable(batch: list[dict])).TN)rrKrr callbacks r! subscribezMonitoringEmitter.subscribevs2 4, , ,   $ $X . . . r#c~ |j|n#t$rYnwxYw|js d|_dSdS)NF)rremove ValueErrorrrTs r! unsubscribezMonitoringEmitter.unsubscribe|sa    $ $X . . . .    D   "!DMMM " "s  **@rEfloatc|dkrdStjd fd }tj|dd}|| dS) zCWait boundedly for queued and in-flight batches to finish dispatch.rNr rcbjdSN)rjoinset)finishedr sr!_wait_for_completionz5MonitoringEmitter.flush.._wait_for_completions# GLLNNN LLNNNNNr#zhermes-monitoring-flushTr;rDr r)rrr?rAwait)r rErcwaiterrbs` @r!flushzMonitoringEmitter.flushs a<< F?$$       !'*      g &&&&&r#Dict[str, int]cv|j|j|jt |jdS)N)queued dispatcheddropped subscribers)rqsizerrrIrrBs r!statszMonitoringEmitter.statss7gmmoo*}t011    r#c|j|j|jdd|_dS)Nr[rDF)rrarr`rrBs r!closezMonitoringEmitter.closes@  < # L  c  * * * r#N)r r r rr$rr rrd)r[)rEr\r r)r rh)__name__ __module__ __qualname____doc__r"r9r0r@rLrVrZrgrorqr#r!r r $sII*. % % % % % %BBBB4 ! ! ! !(((($'''' """"'''''&    r#r Optional[MonitoringEmitter]_EMITTERr cttSt5ttdadddn #1swxYwYtS)z+Return the process-wide monitoring emitter.NFr )ry _EMITTER_LOCKr rwr#r! get_emitterr|s 88  )777H 888888888888888 Os :>>r$rrcHt|dS)z1Module-level convenience: emit via the singleton.N)r|r9)r$s r!r9r9s MMur#emitterct5t4|tur+ tn#t$rYnwxYw|addddS#1swxYwYdS)z Swap the singleton (tests only).N)r{ryrqr5)r~s r!reset_emitter_for_testsrs   G8$;$;          s0A4A AAAAAA)r TelemetryEmitterr|r9r)r r rrr_)r~rxr r)rv __future__rloggingrrr.typingrrr getLoggerrsr6rrJr ry__annotations__rr{r|r9rr__all__rwr#r!rs:*#"""""  &&&&&&&&&&  8 $ $   @@@@@@@@H)-,,,,            %   r#