Added OTel instrumentation and metrics export for sync client

dev_requirements.txt

@@ -30,3 +30,7 @@ redis-entraid==1.0.0
 pybreaker>=1.4.0
 
 xxhash==3.6.0
+
+opentelemetry-api>=1.30.0
+opentelemetry-sdk>=1.30.0
+opentelemetry-exporter-otlp>=1.30.0

pyproject.toml

@@ -52,6 +52,11 @@ jwt = [
 circuit_breaker = [
     "pybreaker>=1.4.0"
 ]
+otel = [
+    "opentelemetry-api>=1.30.0",
+    "opentelemetry-sdk>=1.30.0",
+    "opentelemetry-exporter-otlp-proto-http>=1.30.0",
+]
 
 [project.urls]
 Changes = "https://github.com/redis/redis-py/releases"

redis/_parsers/base.py

@@ -112,7 +112,7 @@ def parse_error(cls, response):
             exception_class = cls.EXCEPTION_CLASSES[error_code]
             if isinstance(exception_class, dict):
                 exception_class = exception_class.get(response, ResponseError)
-            return exception_class(response)
+            return exception_class(response, status_code=error_code)
         return ResponseError(response)
 
     def on_disconnect(self):

redis/cache.py

@@ -4,6 +4,8 @@
 from enum import Enum
 from typing import Any, List, Optional, Union
 
+from redis.observability.attributes import CSCReason
+
 
 class CacheEntryStatus(Enum):
     VALID = "VALID"
@@ -186,9 +188,6 @@ def set(self, entry: CacheEntry) -> bool:
         self._cache[entry.cache_key] = entry
         self._eviction_policy.touch(entry.cache_key)
 
-        if self._cache_config.is_exceeds_max_size(len(self._cache)):
-            self._eviction_policy.evict_next()
-
         return True
 
     def get(self, key: CacheKey) -> Union[CacheEntry, None]:
@@ -248,6 +247,61 @@ def is_cachable(self, key: CacheKey) -> bool:
         return self._cache_config.is_allowed_to_cache(key.command)
 
 
+class CacheProxy(CacheInterface):
+    """
+    Proxy object that wraps cache implementations to enable additional logic on top
+    """
+
+    def __init__(self, cache: CacheInterface):
+        self._cache = cache
+
+    @property
+    def collection(self) -> OrderedDict:
+        return self._cache.collection
+
+    @property
+    def config(self) -> CacheConfigurationInterface:
+        return self._cache.config
+
+    @property
+    def eviction_policy(self) -> EvictionPolicyInterface:
+        return self._cache.eviction_policy
+
+    @property
+    def size(self) -> int:
+        return self._cache.size
+
+    def get(self, key: CacheKey) -> Union[CacheEntry, None]:
+        return self._cache.get(key)
+
+    def set(self, entry: CacheEntry) -> bool:
+        is_set = self._cache.set(entry)
+
+        if self.config.is_exceeds_max_size(self.size):
+            # Lazy import to avoid circular dependency
+            from redis.observability.recorder import record_csc_eviction
+
+            record_csc_eviction(
+                count=1,
+                reason=CSCReason.FULL,
+            )
+            self.eviction_policy.evict_next()
+
+        return is_set
+
+    def delete_by_cache_keys(self, cache_keys: List[CacheKey]) -> List[bool]:
+        return self._cache.delete_by_cache_keys(cache_keys)
+
+    def delete_by_redis_keys(self, redis_keys: List[bytes]) -> List[bool]:
+        return self._cache.delete_by_redis_keys(redis_keys)
+
+    def flush(self) -> int:
+        return self._cache.flush()
+
+    def is_cachable(self, key: CacheKey) -> bool:
+        return self._cache.is_cachable(key)
+
+
 class LRUPolicy(EvictionPolicyInterface):
     def __init__(self):
         self.cache = None
@@ -422,4 +476,4 @@ def __init__(self, cache_config: Optional[CacheConfig] = None):
 
     def get_cache(self) -> CacheInterface:
         cache_class = self._config.get_cache_class()
-        return cache_class(cache_config=self._config)
+        return CacheProxy(cache_class(cache_config=self._config))