yjbMzUdZddlZddlZddlZddlZddlmZddlmZm Z ddl Z ddl m Z ddl mZddlmZerddlmZejeZdd d d d ed edzdefdZdedefdZGddZdaedzed<dedzfdZ d&dedzde eefdzde!de"d e!d!e!d"e!d#edefd$Z#d'd%Z$dS)(a/ Buffered CloudEvents HTTP emitter with retry logic. This module provides a TelemetryEmitter class that: - Buffers events in memory for efficient batching - Sends events as CloudEvents to a configurable HTTP endpoint - Implements exponential backoff retry logic - Supports graceful shutdown with flush N)deque) TYPE_CHECKINGAny)to_json) CloudEvent)HONCHO_VERSION) BaseEventevent_ideventr rater returnct|ttzsdSt|}|dkrdS|dkrdSt|dd}t|tr|r|}n||}n|}tt|dddd z}|t|d zkS) ukTrace-coherent deterministic sampler for high-volume events. `rate` is typed as `object` (rather than `float`) because the caller reads it straight from `settings.TELEMETRY.HIGH_VOLUME_SAMPLE_RATE`, which in tests gets MagicMock'd. A MagicMock comparison against 1.0 raises TypeError, so we validate at the boundary and fall back to passthrough on anything non-numeric. When the event carries a `run_id`, sampling decisions hash on that id — so every event in an agent run either passes or fails the sampler, and join queries downstream don't see half-traces. Events without `run_id` (summarizer, deriver — non-agentic call sites) sample independently per event using the event's deterministic id. Callers that have already computed `event.generate_id()` can pass it as `event_id` to avoid the redundant sha256 hash. T?gFrun_idNbig') isinstanceintfloatgetattrstr generate_id from_bytes _stable_hash)r r r rate_frkeybuckets =/DATA/AppData/hermes/projects/honcho/src/telemetry/emitter.py_should_sampler!s& dC%K ( (t 4[[F }}t }}u UHd + +F&#"6"  !! ^^L--bqb15 9 9F BF C(( ((valuec~ddl}||dS)Nrzutf-8)hashlibsha256encodedigest)r#r%s r rrDs5NNN >>%,,w// 0 0 7 7 9 99r"ceZdZUdZedzed<eeefed<eed<eed<eed<eed<eed <e ed <e e ed <e j ddzed <ejdzed <e ed<e jed<e ed<ee j ded< d&dedzdeeefdzdedededed ed e fdZd'dZd'dZd(dZd'd Zd!ee de fd"Zd'd#Zedefd$Zede fd%ZdS))TelemetryEmittera0Buffered, async CloudEvents emitter with retry logic. This emitter queues events and sends them in batches to reduce network overhead. It supports automatic periodic flushing and exponential backoff retry on failure. Usage: emitter = TelemetryEmitter( endpoint="https://telemetry.example.com/v1/events", headers={"X-API-Key": "..."}, ) await emitter.start() # Queue events (non-blocking) emitter.emit(my_event) # Graceful shutdown await emitter.shutdown() Nendpointheaders batch_sizeflush_intervalflush_threshold max_retriesmax_buffer_sizeenabled_buffer _flush_task_client_running_lock_capacity_warning_active_pending_flush_tasksdr2rTflush_interval_secondsc :||_|pi|_||_||_||_||_||_|o|du|_t||_ d|_ d|_ d|_ tj|_d|_t#|_dS)a&Initialize the telemetry emitter. Args: endpoint: CloudEvents HTTP endpoint URL headers: Optional HTTP headers for authentication batch_size: Maximum events per batch flush_interval_seconds: How often to flush the buffer flush_threshold: Trigger flush when buffer reaches this size max_retries: Maximum retry attempts on failure max_buffer_size: Maximum events to buffer (oldest dropped if exceeded) enabled: Whether emission is enabled N)maxlenF)r+r,r-r.r/r0r1r2rr3r4r5r6asyncioLockr7r8setr9) selfr+r,r-r=r/r0r1r2s r __init__zTelemetryEmitter.__init__os.! }" $4.&.784#7 O444   \^^ (-%$'EE!!!r"rc<K|jstddStjdddi|j|_d|_tj | |_ td|j dS) zqStart the emitter background tasks. Creates the HTTP client and starts the periodic flush task. z4Telemetry emitter disabled (no endpoint or disabled)Ng>@ Content-Typezapplication/cloudevents+json)timeoutr,Tz'Telemetry emitter started, endpoint: %s) r2loggerinfohttpx AsyncClientr,r5r6r@ create_task_periodic_flushr4r+rCs r startzTelemetryEmitter.starts |  KKN O O O F( >,     ".t/C/C/E/EFF =t}MMMMMr"c K|jsdSd|_|j\|jt jt j5|jd{Vdddn #1swxYwYt|j }|rt j |ddid{V| d{V|j &|j d{Vd|_ tddS)u Gracefully shutdown the emitter. Stops the periodic flush task, drains any in-flight threshold flushes, flushes remaining events, and closes the HTTP client. Threshold flushes are spawned from emit() and pop their batch under lock before releasing it for the HTTP send. If we don't await those tasks first, the final flush() can see an empty buffer and return while the in-flight task is still mid-send — closing the HTTP client then orphans that batch. NFreturn_exceptionsTz#Telemetry emitter shutdown complete)r2r6r4cancel contextlibsuppressr@CancelledErrorlistr9gatherflushr5acloserHrI)rCpendings r shutdownzTelemetryEmitter.shutdownsv|  F    '   # # % % %$W%;<< ' '&&&&&&&& ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' t011  C.'BTBB B B B B B B Bjjll < #,%%'' ' ' ' ' ' ' 'DL 9:::::sA++A/2A/r r c *|jsdSddlm}ddlm}d}|dkrt |dd}t|tot|}|s| }t||j j |s*||dS|| }|j j}d |}|rd |d |}||||jd |d |d } |d} t,| d<t/|j|jk} t5| | } | r|d|j| t/|j} |||| t> d|| | |jz }|dkr4|j!s,d|_!t>"d|dz| |jnd|_!t> d| | |j#kr|j$rt> d|  tKj&tKj'|(}|j)*||+|j)j,dS#tZ$rt> dYdSwxYwdSdS)a>Queue an event for emission. Non-blocking. The event is converted to a CloudEvent and added to the buffer. If the buffer is full, the oldest events are dropped. Triggers an immediate flush when buffer reaches flush_threshold. Args: event: The Honcho event to emit Nr)settingsprometheus_metrics high_volumerr ) event_typez/honcho//zhttps://honcho.dev/schemas/z/v)idsourcetypetime dataschemajson)modehoncho_version buffer_fullreason)sizez!Queued event %s (buffer size: %d)g?Tz2Telemetry buffer at %.0f%% capacity (%d/%d events)r:Fz(Event added to emitter (buffer size: %d)z"Triggering flush (buffer size: %d)zEemit() called outside an event loop; deferring flush to periodic task).r2 src.configr] src.telemetry.prometheus.metricsr_ volume_classrrrboolrr! TELEMETRYHIGH_VOLUME_SAMPLE_RATE"record_telemetry_event_sampled_outra NAMESPACEcategory timestamp isoformatschema_version model_dumprlenr3r1rrecord_telemetry_event_droppedappendrecord_telemetry_event_emittedset_telemetry_buffer_sizerHdebugr8warningr/r6r@get_running_looprLrXr9addadd_done_callbackdiscard RuntimeError)rCr r]r_r r has_run_id namespacerd attributesbodywill_drop_oldest cloud_event buffer_sizecapacity_ratio flush_tasks r emitzTelemetryEmitter.emits|  F''''''GGGGGG( $     = 0 0UHd33F#FC00AT&\\J / ,,..!":!  #EE$//11F  ((**H&0 .ENN,,..  ?> >>ENN,<,<>>F$$&&O--//f8H8H8J8JffeNbNbNdNdff & &  %//V/<<!/  t|,,0DD!T22  T  = =] = S S S K((($,'' 99UEUEUEWEW9XXX44+4FFF 8(KPPP %t';; S 0 04-H"S(( -2D ) ?MMM $. . .4= . LL={ K K K (***$0>> )--j999,,T-F-NOOOOO    [  / . . .s2A2M&&$N NcnK|jr|jr|jdS |j4d{V|js dddd{VdSg}|jrct ||jkrK||j|jrt ||jkK dddd{Vn#1d{VswxYwY|sdS||d{V}|rddl m }|j4d{Vt|D]O}t |j|j kr| d|j|P dddd{Vn#1d{VswxYwYtdt |dS)uFlush buffered events to the endpoint. Sends events in batches up to batch_size. Uses exponential backoff retry on failure. Events are returned to the buffer on permanent failure. The lock is held only for buffer mutations (pop batch / restore on failure) — never across the HTTP send. A failing endpoint can spend tens of seconds in retry + backoff; keeping that out of the lock lets concurrent flushers make progress on disjoint batches. NTrr^ send_failedrlz5Failed to send batch of %d events, returned to buffer)r2r3r5r7r|r-r~popleft _send_batchrpr_reversedr1r} appendleftrHr)rCbatchsuccessr_r s r rXzTelemetryEmitter.flushSs=| 4< 4<3G F z 9 9 9 9 9 9 9 9| 9 9 9 9 9 9 9 9 9 9 9 9 9 9+-l9s5zzDO'C'CLL!5!5!7!7888l9s5zzDO'C'C'C  9 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9  ,,U33333333G  K K K K K Kz 3 3 3 3 3 3 3 3&e__33E4<((D,@@@*II#0JL++E2222 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 NNGE     Fs+ CA,C C C  A E<< F FrcpK|j|jdSt|jD]} t |dkrt |d}n2d|D}t j|}|j |j|t |dkrddindd{V}| t d t ||j d S#tj$r/}td |dz|j|Yd}~nDd}~wtj$r/}td |dz|j|Yd}~nd}~wwxYw||jdz krd |z}t%j|d{VdS)zSend a batch of events to the endpoint with retry logic. Args: batch: List of CloudEvents to send Returns: True if successful, False if all retries failed NFrcPg|]#}tjt|$S)rhloadsr).0es r z0TelemetryEmitter._send_batch..s("I"I"Ia4:gajj#9#9"I"I"Ir"rFz"application/cloudevents-batch+json)contentr,z$Sent batch of %d events (status: %d)Tz-HTTP error sending events (attempt %d/%d): %sz0Request error sending events (attempt %d/%d): %s)r5r+ranger0r|rrhdumpsr'postraise_for_statusrHr status_coderJHTTPStatusErrorr RequestErrorr@sleep)rCrattemptr events_jsonresponserdelays r rzTelemetryEmitter._send_batchs@ < 4=#85T-..0 +0 +G* u::??"58,,DD#J"I5"I"I"IK:k2299;;D!%!2!2M u::>>()MNN! "3""))+++ :JJ( tt(   CaK$  %   FaK$  )A---7 mE*********us$CD  F%EF%E??Fc4K|jr tj|jd{V|jr|d{VnC#tj$rYdSt$r%}t d|Yd}~nd}~wwxYw|jdSdS)z5Background task that periodically flushes the buffer.NzError in periodic flush: %s) r6r@rr.r3rXrU ExceptionrH exception)rCrs r rMz TelemetryEmitter._periodic_flushsm C CmD$7888888888<'**,,&&&&&&&)    C C C  !>BBBBBBBB Cm C C C C CsAA B  B 'BB c*t|jS)z-Return the current number of buffered events.)r|r3rNs r rzTelemetryEmitter.buffer_sizes4<   r"c|jS)z&Return whether the emitter is running.)r6rNs r is_runningzTelemetryEmitter.is_runnings }r"NNr:rr;r<rTrN)r r rN)__name__ __module__ __qualname____doc__r__annotations__dictrrrrrrr@TaskrJrKrArBrDrOr[rrXrVrrMpropertyrrrr"r r*r*Js|(Dj #s(^OOO MMM : d#d****   %%%%NNN <""""gl401111 $)-(+!$&*&**&*c3h$&&* &* !& &*  &*&*&*&*&*&*&*PNNNN(%;%;%;%;NB....`>tJ'7>D>>>>@ C C C C!S!!!X!DXr"r*_emitterctS)z*Get the global telemetry emitter instance.)rrr"r get_emitterrs Or"r:rr;r<rTr+r,r-r=r/r0r1r2c Kt||||||||atd{VtS)a)Initialize and start the global telemetry emitter. Args: endpoint: CloudEvents HTTP endpoint URL headers: Optional HTTP headers for authentication batch_size: Maximum events per batch flush_interval_seconds: How often to flush the buffer flush_threshold: Trigger flush when buffer reaches this size max_retries: Maximum retry attempts on failure max_buffer_size: Maximum events to buffer enabled: Whether emission is enabled Returns: The initialized emitter instance r+r,r-r=r/r0r1r2N)r*rrOrs r initialize_emitterrs]6 5''   H ..   Or"c^Kt#td{VdadSdS)z&Shutdown the global telemetry emitter.N)rr[rr"r shutdown_emitterr sC!!!!!!!!!r"rr)%rr@rSrhlogging collectionsrtypingrrrJcloudevents.conversionrcloudevents.httpr src._versionrsrc.telemetry.events.baser getLoggerrrHobjectrrrr!bytesrr*rrrrrrrrrr"r rsa %%%%%%%% ******''''''''''''4333333  8 $ $AE#)#)#) #)$#)36:#) #)#)#)#)L:::::: MMMMMMMMb %) T !(((%,  %)$' &&Dj& #s(^d "&&" &  &  &&&&&&&Rr"