KidBUdZddlmZddlmZmZmZmZddlmZm Z m Z m Z m Z m Z ddlmZmZddlmZddlmZddlmZmZmZerdd lmZdd lmZdd lmZd aeee d <dZ!dZ"edgdd dXde#de$dee#dee%dee#dee&dee'dee%dee%dd fdZ(ddde$dd fd Z) dYd!e#d"e d#e%dd fd$Z*ed%d&'dZd(Z+ed%d&'d)eddd fd*Z,d!e#dd fd+Z-d!e#de$dd fd,Z. d[d-eed.ee&dd fd/Z/d0e#d1e#d2e'dd fd3Z0d!e#dd fd4Z1 d\dee#dee%d5ee#d6ee%d.ee&dee%d7e'dd fd8Z2 d[d9e d:ee#d;ee'dd fd<Z3ed=gd>d d]d?e$d@ee#dAee#d=ee#dd f dBZ4ed=gd>d d[dAee#d=ee#dd fdCZ5de#de%d5e#d6e%d1e#dd f dDZ6 d^dEee fdFZ7dZdGZ8 d^dHed!ee#dd fdIZ9 d^dJe%dKee dd fdLZ:dMe%dd fdNZ;dOdPdQdPdKe dd fdRZdZdVZ?de'fdWZ@y )_aT Simple, clean API for recording observability metrics. This module provides a straightforward interface for Redis core code to record metrics without needing to know about OpenTelemetry internals. Usage in Redis core code: from redis.observability.recorder import record_operation_duration start_time = time.monotonic() # ... execute Redis command ... record_operation_duration( command_name='SET', duration_seconds=time.monotonic() - start_time, server_address='localhost', server_port=6379, db_namespace='0', error=None ) )datetime) TYPE_CHECKINGCallableListOptional)AttributeBuilderConnectionState CSCReason CSCResultGeoFailoverReasonPubSubDirection) CloseReasonRedisMetricsCollector)get_observability_instance)!get_observables_registry_instance)deprecated_argsdeprecated_function str_if_bytes)ConnectionPoolInterface) SyncDatabase) OTelConfigN_metrics_collector csc_itemsconnection_count batch_sizezXThe batch_size argument is no longer used and will be removed in the next major version.z7.2.1) args_to_warnreasonversion command_nameduration_secondsserver_address server_port db_namespaceerror is_blockingretry_attemptsreturnc ttaty tj|||||||||| y#t$rYywxYw)a Record a Redis command execution duration. This is a simple, clean API that Redis core code can call directly. If observability is not enabled, this returns immediately with zero overhead. Args: command_name: Redis command name (e.g., 'GET', 'SET') duration_seconds: Command execution time in seconds server_address: Redis server address server_port: Redis server port db_namespace: Redis database index error: Exception if command failed, None if successful is_blocking: Whether the operation is a blocking command batch_size: Number of commands in batch (for pipelines/transactions) retry_attempts: Number of retry attempts made Example: >>> start = time.monotonic() >>> # ... execute command ... >>> record_operation_duration('SET', time.monotonic() - start, 'localhost', 6379, '0') N) rr r!r"r# error_typenetwork_peer_addressnetwork_peer_portr%r&)r_get_or_create_collectorrecord_operation_duration Exception) rr r!r"r#r$r%rr&s h/home/jay/workspace/scripts/.codegraph-venv/lib/python3.12/site-packages/redis/observability/recorder.pyr-r-2sgR!57  %  44%-)#%!/)#) 5     s9 AAconnection_poolrcttaty tj||y#t$rYywxYw)as Record connection creation time. Args: connection_pool: Connection pool implementation duration_seconds: Time taken to create connection in seconds Example: >>> start = time.monotonic() >>> # ... create connection ... >>> record_connection_create_time('ConnectionPool', time.monotonic() - start) Nr0r )rr,record_connection_create_timer.r2s r/r3r3tsN&!57  %  88+- 9     1 == pool_nameconnection_statecountercttaty tj|||y#t$rYywxYw)a Record a connection count change for a single state. Args: pool_name: Connection pool identifier connection_state: State to update (IDLE or USED) counter: Number to add (positive) or subtract (negative) Example: # New connection created (goes to IDLE first) >>> record_connection_count('pool_abc123', ConnectionState.IDLE, 1) # Acquire from pool (transition) >>> record_connection_count('pool_abc123', ConnectionState.IDLE, -1) >>> record_connection_count('pool_abc123', ConnectionState.USED, 1) # Release to pool (transition) >>> record_connection_count('pool_abc123', ConnectionState.USED, -1) >>> record_connection_count('pool_abc123', ConnectionState.IDLE, 1) # Pool disconnect 5 idle connections >>> record_connection_count('pool_abc123', ConnectionState.IDLE, -5) Nr5r6r7)rr,record_connection_countr.r9s r/r:r:sQ<!57  %  22- 3    2 >>z{Connection count is now tracked via record_connection_count(). This functionality will be removed in the next major versionz7.4.0)rrcht}|yd} |j|y#t$rYywxYw)zB Initialize observable gauge for connection count metric. Nct}|jt}g}|D]}|j||SN)rgetCONNECTION_COUNT_REGISTRY_KEYextend__observables_registry callbacks observationscallbacks r/observable_callbackz2init_connection_count..observable_callbacksH@B(,,-JK  ! ,H    + ,rG)r,init_connection_countr.) collectorrHs r/rKrKsI)*I ''( (     s % 11connection_poolsct}|y ddlmfd}t}|j t |y#t $rYywxYw)zG Add connection pools to connection count observable registry. Nr Observationc|g}D]3}|jD]\}}|j|| 5|S)N attributes)get_connection_countappend)rFr0countrSrPrMs r/connection_count_callbackzBregister_pools_connection_count..connection_count_callbacksUL#3 S)8)M)M)OS%E: '' Ej(QRS S rI)r,opentelemetry.metricsrPrregisterr@r.)rMrLrWrDrPs` @r/register_pools_connection_countrZsW)*I 5  AB%% )+D     s,> A  A c~ttaty tj|y#t$rYywxYw)z Record a connection timeout event. Args: pool_name: Connection pool identifier Example: >>> record_connection_timeout('ConnectionPool') Nr5)rr,record_connection_timeoutr.r\s r/r]r]K!57  %  44 5     0 <<cttaty tj||y#t$rYywxYw)at Record time taken to obtain a connection from the pool. Args: pool_name: Connection pool identifier duration_seconds: Wait time in seconds Example: >>> start = time.monotonic() >>> # ... wait for connection from pool ... >>> record_connection_wait_time('ConnectionPool', time.monotonic() - start) Nr5r )rr,record_connection_wait_timer.ras r/rbrbsN$!57  %  66- 7     r4 close_reasonr)cttaty tj||y#t$rYywxYw)a Record a connection closed event. Args: close_reason: Reason for closing (e.g. 'error', 'application_close') error_type: Error type if closed due to error Example: >>> record_connection_closed('ConnectionPool', 'idle_timeout') Nrcr))rr,record_connection_closedr.res r/rfrf>sN !57  %  33%! 4     r4connection_namemaint_notificationrelaxedcttaty tj|||y#t$rYywxYw)a_ Record a connection timeout relaxation event. Args: connection_name: Connection identifier maint_notification: Maintenance notification type relaxed: True to count up (relaxed), False to count down (unrelaxed) Example: >>> record_connection_relaxed_timeout('localhost:6379_a1b2c3d4', 'MOVING', True) Nrgrhri)rr,!record_connection_relaxed_timeoutr.rks r/rlrl\sQ$!57  %  <<+1 =    r;c~ttaty tj|y#t$rYywxYw)z Record a connection handoff event (e.g., after MOVING notification). Args: pool_name: Connection pool identifier Example: >>> record_connection_handoff('ConnectionPool') Nr\)rr,record_connection_handoffr.r\s r/rnrn}r^r_r*r+ is_internalc ttaty tj|||||||y#t$rYywxYw)a Record error count. Args: server_address: Server address server_port: Server port network_peer_address: Network peer address network_peer_port: Network peer port error_type: Error type (Exception) retry_attempts: Retry attempts is_internal: Whether the error is internal (e.g., timeout, network error) Example: >>> record_error_count('localhost', 6379, 'localhost', 6379, ConnectionError(), 3) Nr!r"r*r+r)r&ro)rr,record_error_countr.rqs r/rrrrs]4!57  %   --)#!5/!)# .     s6 AA directionchannelshardedcttaty|}|t}||jrd} tj |||y#t $rYywxYw)a5 Record a PubSub message (published or received). Args: direction: Message direction ('publish' or 'receive') channel: Pub/Sub channel name sharded: True if sharded Pub/Sub channel Example: >>> record_pubsub_message(PubSubDirection.PUBLISH, 'channel', False) N)rsrtru)rr, _get_confighide_pubsub_channel_namesrecord_pubsub_messager.)rsrtrueffective_channelconfigs r/ryrysz$!57  %    &"B"B $  00% 1    A AA consumer_namez[The consumer_name argument is no longer used and will be removed in the next major version. lag_seconds stream_nameconsumer_groupcttaty|}|t}||jrd} tj |||y#t $rYywxYw)z Record the lag of a streaming message. Args: lag_seconds: Lag in seconds stream_name: Stream name consumer_group: Consumer group name consumer_name: Consumer name Nr~rr)rr,rwhide_stream_namesrecord_streaming_lagr.)r~rrr}effective_stream_namer{s r/rrsz,!57  % (  &":":$( ! //#-) 0    r|c ttaty|sy tjj }t }|duxr |j }t|tr|jD]y\}}|rdn t|}|D]`} | D]Y} | \} } t| } | jd\} } td|t| dz z }tj|||[b{y|D]w}t|d}|rdn|}|dD]Y} | \} } t| } | jd\} } td|t| dz z }tj|||[yy#t$rYywxYw)aQ Record streaming lag from XREAD/XREADGROUP response. Parses the response and calculates lag for each message based on message ID timestamp. Args: response: Response from XREAD/XREADGROUP command consumer_group: Consumer group name (for XREADGROUP) consumer_name: Consumer name (for XREADGROUP) N-girr)rr,rnow timestamprwr isinstancedictitemsrsplitmaxintrr.)responserr}rr{rrstream_messagesrmessagesmessage message_id_rr~ stream_entrys r/"record_streaming_lag_from_responsers.!57  %  - lln&&("$.K63K3K h %080@ , _-D< 3L&!0 H#+ (/ A%1*%= '1'7'7'< 1&)#sS^d5J/J&K *??(3(=+9@    &!)  *<?; 0A{%+A G$+MJ!-j!9J#-#3#3C#8LIq"%c3Y$1F+F"GK&;;$/$9'5<       sCE47A>> record_maint_notification_count('localhost', 6379, 'localhost', 6379, 'MOVING') Nr!r"r*r+rh)rr,record_maint_notification_countr.rs r/rrhsW,!57  %   ::)#!5/1 ;     s4 AAresultc~ttaty tj|y#t$rYywxYw)zm Record a Client Side Caching (CSC) request. Args: result: CSC result ('hit' or 'miss') Nr)rr,record_csc_requestr.rs r/rrsK!57  %  -- .     r_cttatyd} tj|y#t$rYywxYw)z; Initialize observable gauge for CSC items metric. Nct}|jt}g}|D]}|j||Sr>)rr?CSC_ITEMS_REGISTRY_KEYrArBs r/rHz+init_csc_items..observable_callbacksH@B(,,-CD  ! ,H    + ,rIrJ)rr,init_csc_itemsr.)rHs r/rrsP !57  %  ))( *     s 3 ??rGcttatyddlmfd} t }|j t |y#t$rYywxYw)z Adds given callback to CSC items observable registry. Args: callback: Callback function that returns the cache size pool_name: Connection pool name for observability NrrOcLtjgS)Nr\rR)rbuild_csc_attributes)rPrGr5sr/csc_items_callbackz7register_csc_items_callback..csc_items_callbacks(  +@@9U   rI)rr,rXrPrrYrr.)rGr5rrDrPs`` @r/register_csc_items_callbackrsY!57  % 2  @B%%&<>PQ    s A AArVrcttaty tj||y#t$rYywxYw)z Record a Client Side Caching (CSC) eviction. Args: count: Number of evictions reason: Reason for eviction NrVr)rr,record_csc_evictionr.rs r/rrsN!57  %  .. /     r4 bytes_savedc~ttaty tj|y#t$rYywxYw)z Record the number of bytes saved by using Client Side Caching (CSC). Args: bytes_saved: Number of bytes saved Nr)rr,record_csc_network_savedr.rs r/rrsK!57  %  33# 4     r_ fail_fromrfail_tocttaty tj|||y#t$rYywxYw)z Record a geo failover. Args: fail_from: Database failed from fail_to: Database failed to reason: Reason for the failover Nrrr)rr,record_geo_failoverr.rs r/rrsQ!57  %  .. /    r;c< tj}||jjsy|j j t jt j}t ||jS#t$rYyt$rYywxYw)z Get or create the global metrics collector. Returns: RedisMetricsCollector instance if observability is enabled, None otherwise N) rget_provider_managerr{enabled_telemetryget_meter_provider get_meterr METER_NAME METER_VERSION ImportErrorr.)managermeters r/r,r,9s,.CCE ?'.."B"B**,66 ! , ,.C.Q.Q %UGNN;;  s0BAB BBBrcp tj}|y|jS#t$rYywxYw)z Get the OTel configuration from the observability manager. Returns: OTelConfig instance if observability is enabled, None otherwise N)rrr{r.)rs r/rwrwTs<,.CCE ?~~ s) ) 55cday)zM Reset the global collector (used for testing or re-initialization). N)rrIr/reset_collectorrds rIc2t tatduS)zw Check if observability is enabled. Returns: True if metrics are being collected, False otherwise N)rr,rrIr/ is_enabledrls!57 T ))rI)NNNNNNN)r)r'N)NN)NNNNNNT)NNNr>)A__doc__rtypingrrrrredis.observability.attributesrr r r r r redis.observability.metricsrrredis.observability.providersrredis.observability.registryr redis.utilsrrrredis.connectionrredis.multidb.databaserredis.observability.configrr__annotations__rr@strfloatrr.boolr-r3r:rKrZr]rbrfrlrnrrryrrrrrrrrrr,rwrrrrIr/rs*::KDJJJ8357;H23:$ 2 e %)!%"&!%"& $$(: : : SM: # : 3- : I  : $:  : SM:  :  : z .    H* * %* *  * Z C     4 C   45     :    6     B+/&* ;' #   <      B    8%)!%*.'+&*$(* SM* #* #3-*  } * # * SM * *  * ^""% % c]% d^%  % P!" h "&$(#' $ $ #$ SM$ C= $  $  $ N!" h %)#'G SMG C=G  G  G T$ $ $ $  $  $  $ P#' Y  0 > $! ! }!  ! L#'  Y    6    0       <(+@"A6 Xl+   *D *rI