Added stadalone client metrics export

Author: vladvildanov

redis/client.py

@@ -45,7 +45,7 @@
     AfterPubSubConnectionInstantiationEvent,
     AfterSingleConnectionInstantiationEvent,
     ClientType,
-    EventDispatcher,
+    EventDispatcher, AfterCommandExecutionEvent,
 )
 from redis.exceptions import (
     ConnectionError,
@@ -478,7 +478,8 @@ def pipeline(self, transaction=True, shard_hint=None) -> "Pipeline":
         between the client and server.
         """
         return Pipeline(
-            self.connection_pool, self.response_callbacks, transaction, shard_hint
+            self.connection_pool, self.response_callbacks, transaction, shard_hint,
+            event_dispatcher=self._event_dispatcher
         )
 
     def transaction(
@@ -662,16 +663,42 @@ def _execute_command(self, *args, **options):
         command_name = args[0]
         conn = self.connection or pool.get_connection()
 
+        # Start timing for observability
+        start_time = time.monotonic()
+
         if self._single_connection_client:
             self.single_connection_lock.acquire()
         try:
-            return conn.retry.call_with_retry(
+            result = conn.retry.call_with_retry(
                 lambda: self._send_command_parse_response(
                     conn, command_name, *args, **options
                 ),
                 lambda _: self._close_connection(conn),
             )
 
+            self._event_dispatcher.dispatch(
+                AfterCommandExecutionEvent(
+                    command_name=command_name,
+                    duration_seconds=time.monotonic() - start_time,
+                    server_address=conn.host,
+                    server_port=conn.port,
+                    db_namespace=str(conn.db),
+                )
+            )
+            return result
+        except Exception as e:
+            self._event_dispatcher.dispatch(
+                AfterCommandExecutionEvent(
+                    command_name=command_name,
+                    duration_seconds=time.monotonic() - start_time,
+                    server_address=conn.host,
+                    server_port=conn.port,
+                    db_namespace=str(conn.db),
+                    error=e,
+                )
+            )
+            raise
+
         finally:
             if conn and conn.should_reconnect():
                 self._close_connection(conn)
@@ -1385,6 +1412,7 @@ def __init__(
         response_callbacks,
         transaction,
         shard_hint,
+        event_dispatcher: EventDispatcher
     ):
         self.connection_pool = connection_pool
         self.connection: Optional[Connection] = None
@@ -1395,6 +1423,7 @@ def __init__(
         self.command_stack = []
         self.scripts: Set[Script] = set()
         self.explicit_transaction = False
+        self._event_dispatcher = event_dispatcher
 
     def __enter__(self) -> "Pipeline":
         return self
@@ -1501,12 +1530,41 @@ def immediate_execute_command(self, *args, **options):
             conn = self.connection_pool.get_connection()
             self.connection = conn
 
-        return conn.retry.call_with_retry(
-            lambda: self._send_command_parse_response(
-                conn, command_name, *args, **options
-            ),
-            lambda error: self._disconnect_reset_raise_on_watching(conn, error),
-        )
+        # Start timing for observability
+        start_time = time.monotonic()
+
+        try:
+            response = conn.retry.call_with_retry(
+                lambda: self._send_command_parse_response(
+                    conn, command_name, *args, **options
+                ),
+                lambda error: self._disconnect_reset_raise_on_watching(conn, error),
+            )
+
+            self._event_dispatcher.dispatch(
+                AfterCommandExecutionEvent(
+                    command_name=command_name,
+                    duration_seconds=time.monotonic() - start_time,
+                    server_address=conn.host,
+                    server_port=conn.port,
+                    db_namespace=str(conn.db),
+                )
+            )
+
+            return response
+        except Exception as e:
+            self._event_dispatcher.dispatch(
+                AfterCommandExecutionEvent(
+                    command_name=command_name,
+                    duration_seconds=time.monotonic() - start_time,
+                    server_address=conn.host,
+                    server_port=conn.port,
+                    db_namespace=str(conn.db),
+                    error=e,
+                )
+            )
+            raise
+
 
     def pipeline_execute_command(self, *args, **options) -> "Pipeline":
         """
@@ -1679,8 +1737,10 @@ def execute(self, raise_on_error: bool = True) -> List[Any]:
             self.load_scripts()
         if self.transaction or self.explicit_transaction:
             execute = self._execute_transaction
+            operation_name = "MULTI"
         else:
             execute = self._execute_pipeline
+            operation_name = "PIPELINE"
 
         conn = self.connection
         if not conn:
@@ -1689,11 +1749,40 @@ def execute(self, raise_on_error: bool = True) -> List[Any]:
             # back to the pool after we're done
             self.connection = conn
 
+        # Start timing for observability
+        start_time = time.monotonic()
+
         try:
-            return conn.retry.call_with_retry(
+            response = conn.retry.call_with_retry(
                 lambda: execute(conn, stack, raise_on_error),
                 lambda error: self._disconnect_raise_on_watching(conn, error),
             )
+
+            self._event_dispatcher.dispatch(
+                AfterCommandExecutionEvent(
+                    command_name=operation_name,
+                    duration_seconds=time.monotonic() - start_time,
+                    server_address=conn.host,
+                    server_port=conn.port,
+                    db_namespace=str(conn.db),
+                    batch_size=len(stack),
+                )
+            )
+            return response
+        except Exception as e:
+            self._event_dispatcher.dispatch(
+                AfterCommandExecutionEvent(
+                    command_name=operation_name,
+                    duration_seconds=time.monotonic() - start_time,
+                    server_address=conn.host,
+                    server_port=conn.port,
+                    db_namespace=str(conn.db),
+                    error=e,
+                    batch_size=len(stack),
+                )
+            )
+            raise
+
         finally:
             # in reset() the connection is disconnected before returned to the pool if
             # it is marked for reconnect.

redis/event.py

@@ -1,11 +1,13 @@
 import asyncio
 import threading
 from abc import ABC, abstractmethod
+from dataclasses import dataclass
 from enum import Enum
 from typing import Dict, List, Optional, Type, Union
 
 from redis.auth.token import TokenInterface
 from redis.credentials import CredentialProvider, StreamingCredentialProvider
+from redis.observability.recorder import record_operation_duration
 
 
 class EventListenerInterface(ABC):
@@ -90,6 +92,7 @@ def __init__(
             ],
             AfterPubSubConnectionInstantiationEvent: [RegisterReAuthForPubSub()],
             AfterAsyncClusterInstantiationEvent: [RegisterReAuthForAsyncClusterNodes()],
+            AfterCommandExecutionEvent: [ExportOperationDurationMetric()],
             AsyncAfterConnectionReleasedEvent: [
                 AsyncReAuthConnectionListener(),
             ],
@@ -295,6 +298,19 @@ def commands(self) -> tuple:
     def exception(self) -> Exception:
         return self._exception
 
+@dataclass
+class AfterCommandExecutionEvent:
+    """
+    Event fired after command execution.
+    """
+    command_name: str
+    duration_seconds: float
+    server_address: Optional[str] = None
+    server_port: Optional[int] = None
+    db_namespace: Optional[str] = None
+    error: Optional[Exception] = None
+    is_blocking: Optional[bool] = None
+    batch_size: Optional[int] = None
 
 class AsyncOnCommandsFailEvent(OnCommandsFailEvent):
     pass
@@ -466,3 +482,19 @@ def _raise_on_error(self, error: Exception):
 
     async def _raise_on_error_async(self, error: Exception):
         raise EventException(error, self._event)
+
+class ExportOperationDurationMetric(EventListenerInterface):
+    """
+    Listener that exports operation duration metric after command execution.
+    """
+    def listen(self, event: AfterCommandExecutionEvent):
+        record_operation_duration(
+            command_name=event.command_name,
+            duration_seconds=event.duration_seconds,
+            server_address=event.server_address,
+            server_port=event.server_port,
+            db_namespace=event.db_namespace,
+            error=event.error,
+            is_blocking=event.is_blocking,
+            batch_size=event.batch_size,
+        )

redis/exceptions.py

@@ -1,28 +1,52 @@
+from enum import Enum
+
 "Core exceptions raised by the Redis client"
 
 
+class ExceptionType(Enum):
+    NETWORK = 'network'
+    TLS = 'tls'
+    AUTH = 'auth'
+    SERVER = 'server'
+
+
 class RedisError(Exception):
-    pass
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.error_type = ExceptionType.SERVER
+
+    def __repr__(self):
+        return f"{self.error_type.value}:{self.__class__.__name__}"
 
 
 class ConnectionError(RedisError):
-    pass
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.error_type = ExceptionType.NETWORK
 
 
 class TimeoutError(RedisError):
-    pass
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.error_type = ExceptionType.NETWORK
 
 
 class AuthenticationError(ConnectionError):
-    pass
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.error_type = ExceptionType.AUTH
 
 
 class AuthorizationError(ConnectionError):
-    pass
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.error_type = ExceptionType.AUTH
 
 
 class BusyLoadingError(ConnectionError):
-    pass
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.error_type = ExceptionType.NETWORK
 
 
 class InvalidResponse(RedisError):
@@ -70,7 +94,9 @@ class ReadOnlyError(ResponseError):
 
 
 class NoPermissionError(ResponseError):
-    pass
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.error_type = ExceptionType.AUTH
 
 
 class ModuleError(ResponseError):
@@ -84,6 +110,7 @@ class LockError(RedisError, ValueError):
     # This was originally chosen to behave like threading.Lock.
 
     def __init__(self, message=None, lock_name=None):
+        super().__init__(message)
         self.message = message
         self.lock_name = lock_name
 
@@ -106,15 +133,22 @@ class AuthenticationWrongNumberOfArgsError(ResponseError):
     were sent to the AUTH command
     """
 
-    pass
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.error_type = ExceptionType.AUTH
 
 
 class RedisClusterException(Exception):
     """
     Base exception for the RedisCluster client
     """
 
-    pass
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.error_type = ExceptionType.SERVER
+
+    def __repr__(self):
+        return f"{self.error_type.value}:{self.__class__.__name__}"
 
 
 class ClusterError(RedisError):
@@ -123,7 +157,9 @@ class ClusterError(RedisError):
     command execution TTL
     """
 
-    pass
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.error_type = ExceptionType.SERVER
 
 
 class ClusterDownError(ClusterError, ResponseError):
@@ -140,6 +176,7 @@ class ClusterDownError(ClusterError, ResponseError):
     def __init__(self, resp):
         self.args = (resp,)
         self.message = resp
+        self.error_type = ExceptionType.SERVER
 
 
 class AskError(ResponseError):
@@ -160,6 +197,7 @@ class AskError(ResponseError):
 
     def __init__(self, resp):
         """should only redirect to master node"""
+        super().__init__(resp)
         self.args = (resp,)
         self.message = resp
         slot_id, new_node = resp.split(" ")
@@ -176,7 +214,7 @@ class TryAgainError(ResponseError):
     """
 
     def __init__(self, *args, **kwargs):
-        pass
+        super().__init__(*args)
 
 
 class ClusterCrossSlotError(ResponseError):
@@ -188,6 +226,10 @@ class ClusterCrossSlotError(ResponseError):
 
     message = "Keys in request don't hash to the same slot"
 
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.error_type = ExceptionType.SERVER
+
 
 class MovedError(AskError):
     """