Add rediscluster support

docker-compose.yml

@@ -13,7 +13,7 @@ services:
       - "memcached"
       - "redis"
       - "zookeeper"
-
+      - "redis-cluster-node"
   cassandra:
     image: "cassandra:3.11"
     environment:
@@ -25,3 +25,5 @@ services:
     image: "redis:4.0.9"
   zookeeper:
     image: "zookeeper:3.4.10"
+  redis-cluster-node:
+    image: docker.io/grokzen/redis-cluster:6.2.0

docs/api/baseplate/clients/index.rst

@@ -13,6 +13,7 @@ Instrumented Client Libraries
    baseplate.clients.kombu: Client for publishing to queues <kombu>
    baseplate.clients.memcache: Memcached Client <memcache>
    baseplate.clients.redis: Redis Client <redis>
+   baseplate.clients.redis_cluster: Redis Cluster Client <redis_cluster>
    baseplate.clients.requests: Requests (HTTP) Client <requests>
    baseplate.clients.sqlalchemy: SQL Client for relational databases (e.g. PostgreSQL) <sqlalchemy>
    baseplate.clients.thrift: Thrift client for RPC to other backend services <thrift>

docs/api/baseplate/clients/redis_cluster.rst

@@ -0,0 +1,92 @@
+``baseplate.clients.redis_cluster``
+===================================
+
+`Redis`_ is an in-memory data structure store used where speed is necessary but
+complexity is beyond simple key-value operations. (If you're just doing
+caching, prefer :doc:`memcached <memcache>`). `Redis-py-cluster`_ is a Python
+client library that supports interacting with Redis when operating in cluster mode.
+
+.. _`Redis`: https://redis.io/
+.. _`redis-py-cluster`: https://github.com/Grokzen/redis-py
+
+.. automodule:: baseplate.clients.redis_cluster
+
+Example
+-------
+
+To integrate redis-py-cluster with your application, add the appropriate client
+declaration to your context configuration::
+
+   baseplate.configure_context(
+      app_config,
+      {
+         ...
+         "foo": ClusterRedisClient(),
+         ...
+      }
+   )
+
+configure it in your application's configuration file:
+
+.. code-block:: ini
+
+   [app:main]
+
+   ...
+
+
+   # required: what redis instance to connect to
+   foo.url = redis://localhost:6379/0
+
+   # optional: the maximum size of the connection pool
+   foo.max_connections = 99
+
+   # optional: how long to wait for a connection to establish
+   foo.timeout = 3 seconds
+
+   # optional: Whether read requests should be directed to replicas as well
+   #  instead of just the primary
+   foo.read_from_replicas = true
+   ...
+
+
+and then use the attached :py:class:`~redis.Redis`-like object in
+request::
+
+   def my_method(request):
+       request.foo.ping()
+
+Configuration
+-------------
+
+.. autoclass:: ClusterRedisClient
+
+.. autofunction:: cluster_pool_from_config
+
+Classes
+-------
+
+.. autoclass:: ClusterRedisContextFactory
+   :members:
+
+.. autoclass:: MonitoredClusterRedisConnection
+   :members:
+
+Runtime Metrics
+---------------
+
+In addition to request-level metrics reported through spans, this wrapper
+reports connection pool statistics periodically via the :ref:`runtime-metrics`
+system.  All metrics are tagged with ``client``, the name given to
+:py:meth:`~baseplate.Baseplate.configure_context` when registering this context
+factory.
+
+The following metrics are reported:
+
+``runtime.pool.size``
+   The size limit for the connection pool.
+``runtime.pool.in_use``
+   How many connections have been established and are currently checked out and
+   being used.
+
+.. versionadded:: 2.1

requirements-transitive.txt

@@ -56,6 +56,7 @@ pyramid==1.10.5
 python-json-logger==2.0.1
 reddit-cqlmapper==0.3.0
 redis==3.5.3
+redis-py-cluster==2.1.2
 regex==2020.11.13
 requests==2.25.1
 sentry-sdk==0.20.1

requirements.txt

@@ -15,3 +15,4 @@ sphinx-autodoc-typehints==1.11.1
 sphinxcontrib-spelling==7.1.0
 webtest==2.0.35
 wheel==0.36.2
+fakeredis==1.5.0

setup.cfg

@@ -102,6 +102,9 @@ ignore_missing_imports = True
 [mypy-pythonjsonlogger.*]
 ignore_missing_imports = True
 
+[mypy-rediscluster.*]
+ignore_missing_imports = True
+
 [mypy-sqlalchemy.*]
 ignore_missing_imports = True
 

setup.py

@@ -12,6 +12,7 @@
     "memcache": ["pymemcache>=1.3.0,<1.4.4"],
     "pyramid": ["pyramid>=1.9.0,<2.0"],
     "redis": ["redis>=2.10.0,<4.0.0"],
+    "redis-py-cluster": ["redis-py-cluster>=2.0.0,<3.0.0"],
     "refcycle": ["objgraph>=3.0,<4.0"],
     "requests": ["advocate>=1.0.0,<2.0"],
     "sentry": ["sentry-sdk>=0.19,<1.0"],

tests/integration/redis_cluster_tests.py

@@ -0,0 +1,158 @@
+import unittest
+
+try:
+    import rediscluster
+except ImportError:
+    raise unittest.SkipTest("redis-py-cluster is not installed")
+
+from baseplate.lib.config import ConfigurationError
+from baseplate.clients.redis_cluster import cluster_pool_from_config
+
+from baseplate.clients.redis_cluster import ClusterRedisClient
+from baseplate import Baseplate
+from . import TestBaseplateObserver, get_endpoint_or_skip_container
+
+redis_endpoint = get_endpoint_or_skip_container("redis-cluster-node", 7000)
+
+
+# This belongs on the unit tests section but the client class attempts to initialise
+# the list of nodes when being instantiated so it's simpler to test here with a redis
+# cluster available
+class ClusterPoolFromConfigTests(unittest.TestCase):
+    def test_empty_config(self):
+        with self.assertRaises(ConfigurationError):
+            cluster_pool_from_config({})
+
+    def test_basic_url(self):
+        pool = cluster_pool_from_config({"rediscluster.url": f"redis://{redis_endpoint}/0"})
+
+        self.assertEqual(pool.nodes.startup_nodes[0]["host"], "redis-cluster-node")
+        self.assertEqual(pool.nodes.startup_nodes[0]["port"], "7000")
+
+    def test_timeouts(self):
+        pool = cluster_pool_from_config(
+            {
+                "rediscluster.url": f"redis://{redis_endpoint}/0",
+                "rediscluster.timeout": "30 seconds",
+            }
+        )
+
+        self.assertEqual(pool.timeout, 30)
+
+    def test_max_connections(self):
+        pool = cluster_pool_from_config(
+            {
+                "rediscluster.url": f"redis://{redis_endpoint}/0",
+                "rediscluster.max_connections": "300",
+            }
+        )
+
+        self.assertEqual(pool.max_connections, 300)
+
+    def test_max_connections_default(self):
+        # https://github.com/Grokzen/redis-py-cluster/issues/435
+        pool = cluster_pool_from_config({"rediscluster.url": f"redis://{redis_endpoint}/0"})
+
+        self.assertEqual(pool.max_connections, 50)
+
+    def test_kwargs_passthrough(self):
+        pool = cluster_pool_from_config(
+            {"rediscluster.url": f"redis://{redis_endpoint}/0"}, example="present"
+        )
+
+        self.assertEqual(pool.connection_kwargs["example"], "present")
+
+    def test_alternate_prefix(self):
+        pool = cluster_pool_from_config(
+            {"noodle.url": f"redis://{redis_endpoint}/0"}, prefix="noodle."
+        )
+        self.assertEqual(pool.nodes.startup_nodes[0]["host"], "redis-cluster-node")
+        self.assertEqual(pool.nodes.startup_nodes[0]["port"], "7000")
+
+    def test_only_primary_available(self):
+        pool = cluster_pool_from_config({"rediscluster.url": f"redis://{redis_endpoint}/0"})
+        node_list = [pool.get_node_by_slot(slot=1, read_command=False) for _ in range(0, 100)]
+
+        # The primary is on port 7000 so that's the only port we expect to see
+        self.assertTrue(all(node["port"] == 7000 for node in node_list))
+
+    def test_read_from_replicas(self):
+        pool = cluster_pool_from_config({"rediscluster.url": f"redis://{redis_endpoint}/0"})
+
+        node_list = [pool.get_node_by_slot(slot=1, read_command=True) for _ in range(0, 100)]
+
+        # Both replicas and primary are available, so we expect to see some non-primaries here
+        self.assertTrue(any(node["port"] != 7000 for node in node_list))
+
+
+class RedisClusterIntegrationTests(unittest.TestCase):
+    def setUp(self):
+        self.baseplate_observer = TestBaseplateObserver()
+
+        baseplate = Baseplate(
+            {
+                "rediscluster.url": f"redis://{redis_endpoint}/0",
+                "rediscluster.timeout": "1 second",
+                "rediscluster.max_connections": "4",
+            }
+        )
+        baseplate.register(self.baseplate_observer)
+        baseplate.configure_context({"rediscluster": ClusterRedisClient()})
+
+        self.context = baseplate.make_context_object()
+        self.server_span = baseplate.make_server_span(self.context, "test")
+
+    def test_simple_command(self):
+        with self.server_span:
+            result = self.context.rediscluster.ping()
+
+        self.assertTrue(result)
+
+        server_span_observer = self.baseplate_observer.get_only_child()
+        span_observer = server_span_observer.get_only_child()
+        self.assertEqual(span_observer.span.name, "rediscluster.PING")
+        self.assertTrue(span_observer.on_start_called)
+        self.assertTrue(span_observer.on_finish_called)
+        self.assertIsNone(span_observer.on_finish_exc_info)
+
+    def test_error(self):
+        with self.server_span:
+            with self.assertRaises(rediscluster.RedisClusterException):
+                self.context.rediscluster.execute_command("crazycommand")
+
+        server_span_observer = self.baseplate_observer.get_only_child()
+        span_observer = server_span_observer.get_only_child()
+        self.assertTrue(span_observer.on_start_called)
+        self.assertTrue(span_observer.on_finish_called)
+        self.assertIsNotNone(span_observer.on_finish_exc_info)
+
+    def test_lock(self):
+        with self.server_span:
+            with self.context.rediscluster.lock("foo-lock"):
+                pass
+
+        server_span_observer = self.baseplate_observer.get_only_child()
+
+        self.assertGreater(len(server_span_observer.children), 0)
+        for span_observer in server_span_observer.children:
+            self.assertTrue(span_observer.on_start_called)
+            self.assertTrue(span_observer.on_finish_called)
+
+    def test_pipeline(self):
+        with self.server_span:
+            with self.context.rediscluster.pipeline("foo") as pipeline:
+                pipeline.set("foo", "bar")
+                pipeline.get("foo")
+                pipeline.get("foo")
+                pipeline.get("foo")
+                pipeline.get("foo")
+                pipeline.get("foo")
+                pipeline.delete("foo")
+                pipeline.execute()
+
+        server_span_observer = self.baseplate_observer.get_only_child()
+        span_observer = server_span_observer.get_only_child()
+        self.assertEqual(span_observer.span.name, "rediscluster.pipeline_foo")
+        self.assertTrue(span_observer.on_start_called)
+        self.assertTrue(span_observer.on_finish_called)
+        self.assertIsNone(span_observer.on_finish_exc_info)