o d?Z1 @dwdee#dee%dAee#dBee%d6ee&dee%dCe'dd fdDdEZ2 dvdFe dGee#dHee'dd fdIdJZ3edKgdLdd dxdMe$dNee#dOee#dKee#dd f dPdQZ4edKgdLdd dvdOee#dKee#dd fdRdSZ5de#de%dAe#dBe%d:e#dd f dTdUZ6 dydVee fdWdXZ7dudYdZZ8 dyd[ed$ee#dd fd\d]Z9 dyd^e%d_ee dd fd`daZ:dbe%dd fdcddZ;dedfdgdfd_e dd fdhdiZdudodpZ?de'fdqdrZ@d S)zaT 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 CsTtdur tatdur dSztj||||||||||d WdSty)YdSw)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) rrr r!r" error_typenetwork_peer_addressnetwork_peer_portr$r%)r_get_or_create_collectorrecord_operation_duration Exception) rrr r!r"r#r$rr%r-[/root/parts/websockify/install/lib/python3.10/site-packages/redis/observability/recorder.pyr+2s() r+connection_poolrcCDtdur tatdur dSz tj||dWdSty!YdSw)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) Nr/r)rr*record_connection_create_timer,r1r-r-r.r2ts r2 pool_nameconnection_statecountercCFtdur tatdur dSz tj|||dWdSty"YdSw)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) Nr4r5r6)rr*record_connection_countr,r8r-r-r.r9s r9z{Connection count is now tracked via record_connection_count(). This functionality will be removed in the next major versionz7.4.0)rrcCsBt}|dur dSdd}z |j|dWdSty YdSw)zB Initialize observable gauge for connection count metric. NcS.t}|t}g}|D]}||q |SN)rgetCONNECTION_COUNT_REGISTRY_KEYextend__observables_registry callbacks observationscallbackr-r-r.observable_callback  z2init_connection_count..observable_callbackrD)r*init_connection_countr,) collectorrEr-r-r.rHs   rHconnection_poolscsZt}|dur dSzddlmfdd}t}|t|WdSty,YdSw)zG Add connection pools to connection count observable registry. Nr Observationcs6g}D]}|D] \}}|||dq q|S)N attributes)get_connection_countappend)rCr/countrNrLrJr-r.connection_count_callbacks zBregister_pools_connection_count..connection_count_callback)r*opentelemetry.metricsrLrregisterr=r,)rJrIrSrAr-rRr.register_pools_connection_counts   rVcCBtdur tatdur dSz tj|dWdSty YdSw)z Record a connection timeout event. Args: pool_name: Connection pool identifier Example: >>> record_connection_timeout('ConnectionPool') Nr4)rr*record_connection_timeoutr,rXr-r-r.rY rYcCr0)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) Nr4r)rr*record_connection_wait_timer,r[r-r-r.r\s r\ close_reasonr'cCr0)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') Nr]r')rr*record_connection_closedr,r^r-r-r.r_>s r_connection_namemaint_notificationrelaxedcCr7)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) Nr`rarb)rr*!record_connection_relaxed_timeoutr,rcr-r-r.rd\s rdcCrW)z Record a connection handoff event (e.g., after MOVING notification). Args: pool_name: Connection pool identifier Example: >>> record_connection_handoff('ConnectionPool') NrX)rr*record_connection_handoffr,rXr-r-r.re}rZreTr(r) is_internalc CsNtdur tatdur dSztj|||||||dWdSty&YdSw)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%rf)rr*record_error_countr,rgr-r-r.rhs" rh directionchannelshardedcCsjtdur tatdur dS|}|durt}|dur|jrd}z tj|||dWdSty4YdSw)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)rirjrk)rr* _get_confighide_pubsub_channel_namesrecord_pubsub_messager,)rirjrkeffective_channelconfigr-r-r.rns$ rn consumer_namez[The consumer_name argument is no longer used and will be removed in the next major version. lag_seconds stream_nameconsumer_groupcCsjtdur tatdur dS|}|durt}|dur|jrd}z tj|||dWdSty4YdSw)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 Nrrrsrt)rr*rlhide_stream_namesrecord_streaming_lagr,)rrrsrtrqeffective_stream_namerpr-r-r.rws$ rwcCs\tdur tatdur dS|sdSzt}t}|duo!|j}t|trg| D]8\}}|r3dnt |}|D])} | D]$} | \} } t | } | d\} } t d|t | d}tj|||dq=q9q+WdS|D]7}t |d}|rudn|}|dD]$} | \} } t | } | d\} } t d|t | d}tj|||dq{qiWdStyYdSw)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-girurr3)rr*rnow timestamprlrv isinstancedictitemsrsplitmaxintrwr,)responsertrqrzrprvrsstream_messagesrxmessagesmessage message_id_r{rr stream_entryr-r-r."record_streaming_lag_from_responses\      rcCsJtdur tatdur dSz tj|||||dWdSty$YdSw)a Record a maintenance notification count. Args: server_address: Server address server_port: Server port network_peer_address: Network peer address network_peer_port: Network peer port maint_notification: Maintenance notification type (e.g., 'MOVING', 'MIGRATING') Example: >>> record_maint_notification_count('localhost', 6379, 'localhost', 6379, 'MOVING') Nr r!r(r)ra)rr*record_maint_notification_countr,rr-r-r.rhs rresultcCrW)zm Record a Client Side Caching (CSC) request. Args: result: CSC result ('hit' or 'miss') Nr)rr*record_csc_requestr,rr-r-r.r  rcCsJtdur tatdur dSdd}z tj|dWdSty$YdSw)z; Initialize observable gauge for CSC items metric. NcSr:r;)rr<CSC_ITEMS_REGISTRY_KEYr>r?r-r-r.rErFz+init_csc_items..observable_callbackrG)rr*init_csc_itemsr,)rEr-r-r.rs  rrDcsdtdur tatdur dSddlmfdd}z t}|t|WdSty1YdSw)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 NrrKcstjddgS)NrXrM)rbuild_csc_attributesr-rLrDr4r-r.csc_items_callbacks  z7register_csc_items_callback..csc_items_callback)rr*rTrLrrUrr,)rDr4rrAr-rr.register_csc_items_callbacks   rrQrcCr0)z Record a Client Side Caching (CSC) eviction. Args: count: Number of evictions reason: Reason for eviction NrQr)rr*record_csc_evictionr,rr-r-r.rs  r bytes_savedcCrW)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,rr-r-r.rrr fail_fromrfail_tocCr7)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,rr-r-r.rs rcCsjz!t}|dus|jjsWdS|tjtj}t||jWSt y+YdSt y4YdSw)z Get or create the global metrics collector. Returns: RedisMetricsCollector instance if observability is enabled, None otherwise N) rget_provider_managerrpenabled_telemetryget_meter_provider get_meterr METER_NAME METER_VERSION ImportErrorr,)managermeterr-r-r.r*9s   r*rcCs6zt}|dur WdS|jWStyYdSw)z Get the OTel configuration from the observability manager. Returns: OTelConfig instance if observability is enabled, None otherwise N)rrrpr,)rr-r-r.rlTs  rlcCsdadS)zM Reset the global collector (used for testing or re-initialization). N)rr-r-r-r.reset_collectordsrcCstdurtatduS)zw Check if observability is enabled. Returns: True if metrics are being collected, False otherwise N)rr*r-r-r-r. is_enabledls r)NNNNNNN)r3)r&N)NN)NNNNNNT)NNNr;)A__doc__rtypingrrrrredis.observability.attributesrrr r r r redis.observability.metricsr rredis.observability.providersrredis.observability.registryr redis.utilsrrrredis.connectionrredis.multidb.databaserredis.observability.configrr__annotations__rr=strfloatrr,boolr+r2r9rHrVrYr\r_rdrerhrnrwrrrrrrrrr*rlrrr-r-r-r.s         = $ -    !  !  / ( ' J (  &