tPj$H UdZddlZddlZddlZddlZddlmZddlmZddl m Z ddl m Z ddl mZmZddlmZdd lmZdd lmZdd lmZmZdd lmZmZmZmZmZmZm Z m!Z!m"Z"m#Z#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z)m*Z*m+Z+m,Z,dd l-m.Z.e/ddhZ0e e1d<de2dzde2fdZ3de.de*fdZ4de5e*dfde5e'dffdZ6de5e*dfde7de7de5e)dffdZ8GddeZ9y)a New Relic Metric API Integration - sends per-team cost/usage metrics to /metric/v1 NR Reference API: https://docs.newrelic.com/docs/data-apis/ingest-apis/metric-api/introduction-metric-api/ `async_log_success_event` / `async_log_failure_event` queue one record per request; at flush the queue is aggregated by (team, model group, model, provider, status) into count/summary metrics. `interval.ms` is the real window between flushes, computed at flush time. Team-scoped by construction: the ingest key is injected explicitly and there is deliberately no environment-variable fallback, so a team's metrics are never sent with the proxy operator's credentials (mirrors ``allow_env_credentials=False`` on the Datadog team logger). Error policy on flush: 4xx drops the batch (a retry would fail identically; 403 is a permanent credential failure), 5xx/network re-queues capped at ``max_queue_size`` records with the oldest dropped. For batching specific details see CustomBatchLogger class N)Mapping)ceil)MappingProxyType)Final)HTTPStatusErrorResponse)verbose_logger)CustomBatchLogger) safe_dumps)get_async_httpx_clienthttpxSpecialProvider)NEWRELIC_DEFAULT_REGION!NEWRELIC_METRIC_ATTRIBUTE_MAX_LEN!NEWRELIC_METRIC_COMPLETION_TOKENSNEWRELIC_METRIC_COST_USD"NEWRELIC_METRIC_ENDPOINT_BY_REGIONNEWRELIC_METRIC_PROMPT_TOKENS#NEWRELIC_METRIC_REQUEST_DURATION_MSNEWRELIC_METRIC_REQUESTSNEWRELIC_METRIC_TOTAL_TOKENSNEWRELIC_METRICS_MAX_BATCH_SIZE!NEWRELIC_METRICS_MAX_DRAIN_PASSES%NEWRELIC_METRICS_MAX_RETRY_QUEUE_SIZENewRelicCountMetricNewRelicMetricNewRelicMetricCommonNewRelicMetricEnvelopeNewRelicMetricRecordNewRelicSummaryMetricNewRelicSummaryValue)StandardLoggingPayloadii_RETRYABLE_CLIENT_STATUSESnewrelic_regionreturnc |s ttStj|j}|@t j d|dj ttttS|S)Nz^New Relic: unknown newrelic_region %r; supported regions: %s. Using the default (US) endpoint.z, )rrgetlowerr warningjoinsorted)r#endpoints s/home/oem/CyberBrain26-unified/.venv/lib/python3.12/site-packages/litellm/integrations/newrelic/newrelic_metrics.py resolve_newrelic_metric_endpointr-Esi 12IJJ8<<_=R=R=TUH l  IIf?@ A 22IJJ Ostandard_logging_objectc|jd}|r$|jdxs|jdndxsd}|r$|jdxs|jdndxsd}t|||jdxsd|jdxsd|jd xsdt|jd xsd t|jd xsd t |jdxsdt |jdxsdt |jdxsdt|jdxsd dz S)Nmetadatauser_api_key_team_idteam_iduser_api_key_team_alias team_alias model_groupmodelcustom_llm_providerstatussuccess response_costg prompt_tokensrcompletion_tokens total_tokens response_timeg@@) r3r6r7r8r9r:r<r=r>r? duration_ms)r&rstrfloatint)r/r1r3r6s r,_metric_record_from_payloadrESsU-11*=H[cx||$:;Vx||I?VimtrtGS[/ 0 NHLL4Nae  +// >D"%))'28b3778MNTRT*..x8EIF377HOCP155oFK!L599:MNSRST044^DIJ155oFM#NQWW  r.bucket_records.c |d}d|jfd|jfd|jfd|jfd|jfd|j ffDcic]\}}|r ||dt c}}td|D}ttt|fttd |Dftttd |Dftttd |Dftttd |Dff}tfd |D}t!t"dt%t|t|t'|t)|}g||Scc}}w)Nrr3r6r7r8r9r:c34K|]}|jywN)rA.0records r, z"_bucket_metrics..vsMFV//Mc34K|]}|jywrI)r<rJs r,rMz"_bucket_metrics..ys&Yv';';&YrNc34K|]}|jywrI)r=rJs r,rMz"_bucket_metrics..zs1d6&2F2F1drNc34K|]}|jywrI)r>rJs r,rMz"_bucket_metrics..{s5lSYf6N6N5lrNc34K|]}|jywrI)r?rJs r,rMz"_bucket_metrics..|s0b1D1D0brNc3BK|]\}}t|d|yw)countnametypevalue attributesN)r)rKrVrXrYs r,rMz"_bucket_metrics..~s+=]h]achG5ZXX=summary)rTsumminmaxrU)r3r6r7r8r9r:rtuplerrClenrr\rrrrrr r]r^) rFfirstkeyrX durationscounts count_metricssummary_metricrYs @r,_bucket_metricsrghs!!$E & 5++ , E-- . ekk " "E$=$= > u|| $   , C  U55 66 ,JMnMMI !5^)<#=> !3&Y.&Y#YZ &c1dUc1d.d(ef *E#5l]k5l2l,mn %uS0bSa0b-b'cd 4F8==lr=8M2 0 "i.III   N ,] +N ++E ,sE7records window_startnowctdt||z dz}ttj dD}tfd|D}t|dz|d}t ||fS)zEAggregates records into one Metric API envelope for the flush window.ic34K|]}|jywrI bucket_keyrJs r,rMz'build_metric_payload..s,U6V->->,UrNc3hK|](ttfdDD]}|*yw)c3BK|]}|jk(s|ywrIrn)rKrLrbs r,rMz1build_metric_payload...s+evFL]L]adLdF+esN)rgr_)rKmetricrbrhs @r,rMz'build_metric_payload..s> %e+e+e&ef  s.2) timestampz interval.ms)commonmetrics)r^rDr_dictfromkeysr)rhrirj interval_ms bucket_keysrurts` r,build_metric_payloadrzs{QS<%74$? @AKt}},UW,UUVKG ,-"+F #&' B DDr.ceZdZ ddededzddffd ZddZddZddZdd Zdd Z dd Z dd Z d e dzddfdZ ddZddedzdefdZdeedfdedefdZdeedfddfdZdeedfdefdZxZS)NewRelicMetricsLoggerNnewrelic_api_keyr#r$c|s td||_t||_t t j |_d|_tj|_ tj|jtj|_t|A|jt"t$y)Nzvnewrelic_api_key is required for NewRelicMetricsLogger; team-scoped metrics never fall back to environment credentials) llm_providerF) flush_lock batch_sizemax_queue_size) ValueErrorr}r-metric_api_urlr r LoggingCallback async_client_stoppedasyncioLock _drain_lock create_taskperiodic_flushrsuper__init__rr)selfr}r# __class__s r,rzNewRelicMetricsLogger.__init__s  Q (8%Eo%V2@T@d@de# "<<>D//12!,,. 6@  r.cd|_ tjj|j y#t $rt jddYywxYw)aEnds the periodic flush loop; called on DynamicLoggingCache eviction. Schedules one final drain of anything still queued, so eviction never silently discards records. Guarded so it can never raise into the cache's eviction path. Tz;New Relic Metrics: could not schedule final drain on stop())exc_infoN)rrget_running_loopr _final_drain Exceptionr debugrs r,stopzNewRelicMetricsLogger.stopsT  o  $ $ & 2 243D3D3F G o  !^im n os1; AAc:K|j4d{t|j}ttD]e}|j d{|jsdddd{y|tdz ksFt jd|zd{g|j4d{td|Dtfd|jD}t|jt|z }|rtjd|tt||jdddddd{dddd{y7M77777!#1d{7swY1xYw7(#1d{7swYyxYww)aDeliver everything queued on a stopped logger, or drop it with a log. A stopped logger has no periodic loop left, so every post-stop path funnels through here. ``_drain_lock`` serializes drains: a callback that appends and starts its own drain queues behind the running one instead of racing it. Each pass attempts the whole current queue in ``batch_size`` chunks, unlike the periodic path it does not stop at the first failing chunk, so a persistently failing head never starves the tail. Only after ``_MAX_DRAIN_PASSES`` against a permanently failing destination is the remainder dropped, and then only the records that were queued when this drain began, so every dropped record got the full retry budget: a record a callback appended mid-drain is not in that snapshot, so it is left for its own serialized drain rather than dropped after fewer attempts, and is never stranded. Nrlc32K|]}t|ywrIidrJs r,rMz:NewRelicMetricsLogger._drain_with_retry..s,PFRZ,Psc3>K|]}t|vs|ywrIr)rKrL tried_idss r,rMz:NewRelicMetricsLogger._drain_with_retry..s (lFPRSYPZbkPk(lsz.s& 9>GEEDOO3 4 rZrc3<K|]\}}|rdn|D]}|yw)Nr)rKchunkokrLs r,rMz:NewRelicMetricsLogger._drain_flush_once..s(qbd\^joqQWfqfqs) rr_rlast_flush_timetimerr`r_classify_and_sendzip_requeue)rrichunksr deliveredfailedrs` @r,rz'NewRelicMetricsLogger._drain_flush_onces?? " ""4>>2G"&"6"6L#'99;D q!  " "   BG3w? !22 2MMM        N     sC>CC>>C) C> C!!C>&AC)5C#6C);C)< C>C%C> C) C>C'C>!C>#C)%C>'C>)C;/C2 0C;7C>ricK|jsyt|jd|j}|jdt|=|j |||n |j d{}|s|j ||S7w)zSends the oldest ``batch_size`` records only, so a queue grown past that by re-queues cannot breach the Metric API data point cap in one request. Returns False once a chunk fails and is re-queued, so the caller stops.FN)rr_rr`rrr)rri batch_to_sendrs r,rz&NewRelicMetricsLogger.async_send_batch4s~~AFt~~VgX\XgXgGhAi NN/S// 0!%!8!8 <+C<I]I]"   MM- (  sA+B -B.B batch.c*Kt||tj} |j|d{j}d|cxkrdkryd|cxkrd kr4nn1|tvr)tjd ||d k(rd nd t|ytjd|t|y7{#t$r }|j j}Yd}~d}~wt $r*}tjdt||Yd}~yd}~wwxYww)a!Send one chunk and classify the outcome, never touching the queue. Returns True when the batch is done with (delivered on any 2xx, or a 4xx a retry would only repeat, 403 being a permanent bad-key rejection), and False when a 5xx or network error means the caller should re-queue it. ``AsyncHTTPHandler.post`` raises ``HTTPStatusError`` on any non-2xx, so a 4xx never returns a response here; the status is read off the raised error to keep the client-error path (drop) distinct from 5xx (retry).)rhrirjNzDNew Relic Metrics: network error sending %s records, will retry - %sFi,Tiiz=New Relic Metrics: %s from Metric API%s, dropping %s records.izC (permanent credential failure: invalid or revoked team ingest key)r4z>k  & 3   & 3 61K#K  " "OY_cfYfUlnE    J  J ;? ,ZZ++F   " "VE     sQ!DB6B4B6A-D4B66 D?CD D& D D DDc||jddt|j|jz }|dkDr1|jd|=tjd|j|yy)zPrepends ``batch`` in place (never by assignment: records appended by concurrent requests during the flush await must survive), keeping chronological order so the cap drops the oldest records first.NrzUNew Relic Metrics: retry queue exceeded max_queue_size=%s; dropped %s oldest records.)rr`rr r()rroverflows r,rzNewRelicMetricsLogger._requeuepsf#rdnn-0C0CC a<yy)  " "g##  r.rcKtjt|jd}t dd|j d}|j j|j||d{S7w)Nzutf-8zapplication/jsongzip)z Content-TypezContent-EncodingzApi-Key)urldataheaders) rcompressr encoderr}rpostr)rrcompressed_datars r,rz0NewRelicMetricsLogger.async_send_compressed_data~s{!%z'/B/I/I'/R!S,< 2$*00 - &&++## ,    sA2A;4A95A;rI)r$N)__name__ __module__ __qualname__rBrrrrrrrrr!rrrCboolrr_rrrrrr __classcell__)rs@r,r|r|s'+  t    . o"4H",'"j j %>TW[>[ %`d % 54<4")e4H#4M.N)^c)hl)V e$8#$=> 4   >TVY>Y8Z  _g  r.r|):__doc__rrrrcollections.abcrmathrtypesrtypingrhttpxrrlitellm._loggingr (litellm.integrations.custom_batch_loggerr *litellm.litellm_core_utils.safe_json_dumpsr &litellm.llms.custom_httpx.http_handlerr r #litellm.types.integrations.newrelicrrrrrrrrrrrrrrrrrrr litellm.types.utilsr!rr"__annotations__rBr-rEr_rgrCrzr|rr.r,rs, #"++FA*7%.sCj$9E9 cDj S 9OTh*$,E*>*C$D$,~_bObIc$,NE ', -EE  E  !3 &' E*g -g r.