Added caching support for Sentinel

Author: vladvildanov

redis/connection.py

@@ -729,6 +729,8 @@ def __init__(self, conn: ConnectionInterface, cache: Cache, conf: CacheConfigura
         self.pid = os.getpid()
         self._conn = conn
         self.retry = self._conn.retry
+        self.host = self._conn.host
+        self.port = self._conn.port
         self._cache = cache
         self._conf = conf
         self._current_command_hash = None
@@ -770,7 +772,7 @@ def send_packed_command(self, command, check_health=True):
     def send_command(self, *args, **kwargs):
         self._process_pending_invalidations()
 
-        # If command is write command or not allowed to cache skip it.
+        # If command is write command or not allowed to cache, transfer control to the actual connection.
         if not self._conf.is_allowed_to_cache(args[0]):
             self._current_command_hash = None
             self._current_command_keys = None

redis/sentinel.py

@@ -229,6 +229,7 @@ def __init__(
         sentinels,
         min_other_sentinels=0,
         sentinel_kwargs=None,
+        force_master_ip=None,
         **connection_kwargs,
     ):
         # if sentinel_kwargs isn't defined, use the socket_* options from
@@ -245,6 +246,7 @@ def __init__(
         ]
         self.min_other_sentinels = min_other_sentinels
         self.connection_kwargs = connection_kwargs
+        self._force_master_ip = force_master_ip
 
     def execute_command(self, *args, **kwargs):
         """
@@ -304,7 +306,9 @@ def discover_master(self, service_name):
                     sentinel,
                     self.sentinels[0],
                 )
-                return state["ip"], state["port"]
+
+                ip = self._force_master_ip if self._force_master_ip is not None else state["ip"]
+                return ip, state["port"]
 
         error_info = ""
         if len(collected_errors) > 0:

tests/conftest.py

@@ -410,31 +410,40 @@ def sslclient(request):
 
 
 @pytest.fixture()
-def sentinel_setup(cache, request):
+def sentinel_setup(request):
     sentinel_ips = request.config.getoption("--sentinels")
     sentinel_endpoints = [
         (ip.strip(), int(port.strip()))
         for ip, port in (endpoint.split(":") for endpoint in sentinel_ips.split(","))
     ]
     kwargs = request.param.get("kwargs", {}) if hasattr(request, "param") else {}
+    use_cache = request.param.get("use_cache", False)
+    cache = request.param.get("cache", None)
+    cache_size = request.param.get("cache_size", 128)
+    cache_ttl = request.param.get("cache_ttl", 300)
+    force_master_ip = request.param.get("force_master_ip", None)
     sentinel = Sentinel(
         sentinel_endpoints,
+        force_master_ip=force_master_ip,
         socket_timeout=0.1,
-        use_cache=cache,
+        use_cache=use_cache,
         cache=cache,
+        cache_ttl=cache_ttl,
+        cache_size=cache_size,
         protocol=3,
         **kwargs,
     )
-    yield sentinel
+    yield sentinel, cache
     for s in sentinel.sentinels:
         s.close()
 
 
 @pytest.fixture()
 def master(request, sentinel_setup):
     master_service = request.config.getoption("--master-service")
-    master = sentinel_setup.master_for(master_service)
-    yield master
+    sentinel, cache = sentinel_setup
+    master = sentinel.master_for(master_service)
+    yield master, cache
     master.close()
 
 

tests/test_cache.py

@@ -5,7 +5,6 @@
 from cachetools import TTLCache, LRUCache, LFUCache
 
 import redis
-from redis import Redis, RedisCluster
 from redis.utils import HIREDIS_AVAILABLE
 from tests.conftest import _get_client
 
@@ -113,8 +112,6 @@ def test_health_check_invalidate_cache_multithreaded(self, r, r2, cache):
         threading.Thread(target=r2.set, args=("bar", "bar")).start()
         # Wait for health check
         time.sleep(2)
-        # Trigger object destructor to shutdown health check thread
-        del r
         # Make sure that value was invalidated
         assert cache.get(("GET", "foo")) is None
         assert cache.get(("GET", "bar")) is None
@@ -492,17 +489,92 @@ def test_cache_flushed_on_server_flush(self, r, cache):
 @pytest.mark.skipif(HIREDIS_AVAILABLE, reason="PythonParser only")
 @pytest.mark.onlynoncluster
 class TestSentinelCache:
-    def test_get_from_cache(self, cache, master):
+    @pytest.mark.parametrize(
+        "sentinel_setup",
+        [{"cache": LRUCache(maxsize=128), "use_cache": True, "force_master_ip": "localhost"}],
+        indirect=True,
+    )
+    @pytest.mark.onlynoncluster
+    def test_get_from_cache(self, master, cache):
+        master, cache = master
         master.set("foo", "bar")
         # get key from redis and save in local cache
         assert master.get("foo") == b"bar"
         # get key from local cache
         assert cache.get(("GET", "foo")) == b"bar"
         # change key in redis (cause invalidation)
         master.set("foo", "barbar")
-        # send any command to redis (process invalidation in background)
-        master.ping()
-        # the command is not in the local cache anymore
-        assert cache.get(("GET", "foo")) is None
         # get key from redis
         assert master.get("foo") == b"barbar"
+        # Make sure that new value was cached
+        assert cache.get(("GET", "foo")) == b"barbar"
+
+    @pytest.mark.parametrize(
+        "sentinel_setup",
+        [{"cache": LRUCache(maxsize=128), "use_cache": True, "force_master_ip": "localhost"}],
+        indirect=True,
+    )
+    @pytest.mark.onlynoncluster
+    def test_get_from_cache_multithreaded(self, master, cache):
+        master, cache = master
+        # Running commands over two threads
+        threading.Thread(target=set_get, args=(master, "foo", "bar")).start()
+        threading.Thread(target=set_get, args=(master, "bar", "foo")).start()
+
+        # Wait for command execution to be finished
+        time.sleep(0.1)
+
+        # Make sure that both values was cached.
+        assert cache.get(("GET", "foo")) == b"bar"
+        assert cache.get(("GET", "bar")) == b"foo"
+
+        # Running commands over two threads
+        threading.Thread(target=set_get, args=(master, "foo", "baz")).start()
+        threading.Thread(target=set_get, args=(master, "bar", "bar")).start()
+
+        # Wait for command execution to be finished
+        time.sleep(0.1)
+
+        # Make sure that new values was cached.
+        assert cache.get(("GET", "foo")) == b"baz"
+        assert cache.get(("GET", "bar")) == b"bar"
+
+    @pytest.mark.parametrize(
+        "sentinel_setup",
+        [{"cache": LRUCache(maxsize=128), "use_cache": True, "force_master_ip": "localhost"}],
+        indirect=True,
+    )
+    @pytest.mark.onlynoncluster
+    def test_health_check_invalidate_cache(self, master, cache):
+        master, cache = master
+        # add key to redis
+        master.set("foo", "bar")
+        # get key from redis and save in local cache
+        assert master.get("foo") == b"bar"
+        # get key from local cache
+        assert cache.get(("GET", "foo")) == b"bar"
+        # change key in redis (cause invalidation)
+        master.set("foo", "barbar")
+        # Wait for health check
+        time.sleep(2)
+        # Make sure that value was invalidated
+        assert cache.get(("GET", "foo")) is None
+
+    @pytest.mark.parametrize(
+        "sentinel_setup",
+        [{"cache": LRUCache(maxsize=128), "use_cache": True, "force_master_ip": "localhost"}],
+        indirect=True,
+    )
+    @pytest.mark.onlynoncluster
+    def test_cache_clears_on_disconnect(self, master, cache):
+        master, cache = master
+        # add key to redis
+        master.set("foo", "bar")
+        # get key from redis and save in local cache
+        assert master.get("foo") == b"bar"
+        # get key from local cache
+        assert cache.get(("GET", "foo")) == b"bar"
+        # Force disconnection
+        master.connection_pool.get_connection('_').disconnect()
+        # Make sure cache is empty
+        assert cache.currsize == 0