redis-clustering support

Author: joshuay03

.github/workflows/ci.yml

@@ -14,6 +14,7 @@ jobs:
       matrix:
         ruby: ['2.3', '2.4', '2.5', '2.6', '2.7', '3.0', '3.1', '3.2', '3.3', '3.4', head, jruby-head, truffleruby-head]
         redis: ['4', '5']
+        redis_cluster: [false, true]
         search: [
           ['opensearch-ruby:2', 'opensearchproject/opensearch:2'],
           ['opensearch-ruby:3', 'opensearchproject/opensearch:3'],
@@ -26,6 +27,23 @@ jobs:
             redis: '5'
           - ruby: '2.4'
             redis: '5'
+          # redis-clustering first release is 5.x
+          - redis: '4'
+            redis_cluster: true
+          # redis-clustering 5.x requires ruby >= 2.7
+          - ruby: '2.3'
+            redis_cluster: true
+          - ruby: '2.4'
+            redis_cluster: true
+          - ruby: '2.5'
+            redis_cluster: true
+          - ruby: '2.6'
+            redis_cluster: true
+          # our usage of redis-cluster-client suffers from https://bugs.ruby-lang.org/issues/18991 in ruby <= 3.0
+          - ruby: '2.7'
+            redis_cluster: true
+          - ruby: '3.0'
+            redis_cluster: true
           # opensearch-ruby 2.x requires ruby >= 2.4
           - ruby: '2.3'
             search: ['opensearch-ruby:2', 'opensearchproject/opensearch:2']
@@ -49,10 +67,6 @@ jobs:
           - ruby: '2.5'
             search: ['elasticsearch:9', 'elasticsearch:9.0.2']
     services:
-      redis:
-        image: redis
-        ports:
-          - 6379:6379
       search:
         image: ${{ matrix.search[1] }}
         ports:
@@ -73,6 +87,7 @@ jobs:
 
     env:
       REDIS_VERSION: ${{ matrix.redis }}
+      REDIS_CLUSTER: ${{ matrix.redis_cluster && 'true' || '' }}
       SEARCH_GEM: ${{ matrix.search[0] }}
 
     steps:
@@ -81,6 +96,14 @@ jobs:
         with:
           ruby-version: ${{ matrix.ruby }}
           bundler-cache: true
+      - name: Start Redis (single instance)
+        if: ${{ !matrix.redis_cluster }}
+        run: |
+          docker run -d --name redis -p 6379:6379 redis
+      - name: Start Redis Cluster
+        if: ${{ matrix.redis_cluster }}
+        run: |
+          docker compose -f docker-compose.redis-cluster.yml up -d
       - name: start MySQL
         run: sudo /etc/init.d/mysql start
       - run: bundle exec rspec --format doc

Gemfile

@@ -30,12 +30,20 @@ if Gem::Version.new(RUBY_VERSION) >= Gem::Version.new('2.6')
   gem 'simplecov-cobertura', '~> 2.1'
 end
 
-if ENV['REDIS_VERSION']
-  gem 'redis', "~> #{ENV['REDIS_VERSION']}"
+if (redis_version = ENV.fetch('REDIS_VERSION', nil))
+  gem 'redis', "~> #{redis_version}"
 end
 
-if ENV['SEARCH_GEM']
-  name, version = ENV['SEARCH_GEM'].split(':')
+if redis_version
+  if ENV.fetch('REDIS_CLUSTER', nil) == 'true'
+    gem 'redis-clustering', "~> #{redis_version}"
+  end
+elsif Gem::Version.new(RUBY_VERSION) >= Gem::Version.new('2.7')
+  gem 'redis-clustering' # rubocop:disable Bundler/DuplicatedGem
+end
+
+if (search_gem = ENV.fetch('SEARCH_GEM', nil))
+  name, version = search_gem.split(':')
   gem name, "~> #{version}"
 else
   gem 'opensearch-ruby'

docker-compose.redis-cluster.yml

@@ -0,0 +1,72 @@
+services:
+  redis-cluster-node-0:
+    image: docker.io/bitnami/redis-cluster:latest
+    volumes:
+      - redis-cluster_data-0:/bitnami/redis/data
+    environment:
+      - "ALLOW_EMPTY_PASSWORD=yes"
+      - "REDIS_NODES=redis-cluster-node-0 redis-cluster-node-1 redis-cluster-node-2 redis-cluster-node-3 redis-cluster-node-4 redis-cluster-node-5"
+
+  redis-cluster-node-1:
+    image: docker.io/bitnami/redis-cluster:latest
+    volumes:
+      - redis-cluster_data-1:/bitnami/redis/data
+    environment:
+      - "ALLOW_EMPTY_PASSWORD=yes"
+      - "REDIS_NODES=redis-cluster-node-0 redis-cluster-node-1 redis-cluster-node-2 redis-cluster-node-3 redis-cluster-node-4 redis-cluster-node-5"
+
+  redis-cluster-node-2:
+    image: docker.io/bitnami/redis-cluster:latest
+    volumes:
+      - redis-cluster_data-2:/bitnami/redis/data
+    environment:
+      - "ALLOW_EMPTY_PASSWORD=yes"
+      - "REDIS_NODES=redis-cluster-node-0 redis-cluster-node-1 redis-cluster-node-2 redis-cluster-node-3 redis-cluster-node-4 redis-cluster-node-5"
+
+  redis-cluster-node-3:
+    image: docker.io/bitnami/redis-cluster:latest
+    volumes:
+      - redis-cluster_data-3:/bitnami/redis/data
+    environment:
+      - "ALLOW_EMPTY_PASSWORD=yes"
+      - "REDIS_NODES=redis-cluster-node-0 redis-cluster-node-1 redis-cluster-node-2 redis-cluster-node-3 redis-cluster-node-4 redis-cluster-node-5"
+
+  redis-cluster-node-4:
+    image: docker.io/bitnami/redis-cluster:latest
+    volumes:
+      - redis-cluster_data-4:/bitnami/redis/data
+    environment:
+      - "ALLOW_EMPTY_PASSWORD=yes"
+      - "REDIS_NODES=redis-cluster-node-0 redis-cluster-node-1 redis-cluster-node-2 redis-cluster-node-3 redis-cluster-node-4 redis-cluster-node-5"
+
+  redis-cluster-node-5:
+    image: docker.io/bitnami/redis-cluster:latest
+    volumes:
+      - redis-cluster_data-5:/bitnami/redis/data
+    depends_on:
+      - redis-cluster-node-0
+      - redis-cluster-node-1
+      - redis-cluster-node-2
+      - redis-cluster-node-3
+      - redis-cluster-node-4
+    environment:
+      - "ALLOW_EMPTY_PASSWORD=yes"
+      - "REDIS_CLUSTER_REPLICAS=1"
+      - "REDIS_NODES=redis-cluster-node-0 redis-cluster-node-1 redis-cluster-node-2 redis-cluster-node-3 redis-cluster-node-4 redis-cluster-node-5"
+      - "REDIS_CLUSTER_CREATOR=yes"
+    ports:
+      - "6379:6379"
+
+volumes:
+  redis-cluster_data-0:
+    driver: local
+  redis-cluster_data-1:
+    driver: local
+  redis-cluster_data-2:
+    driver: local
+  redis-cluster_data-3:
+    driver: local
+  redis-cluster_data-4:
+    driver: local
+  redis-cluster_data-5:
+    driver: local

lib/faulty/patch/redis.rb

@@ -1,6 +1,11 @@
 # frozen_string_literal: true
 
 require 'redis'
+begin
+  require 'redis-clustering'
+rescue LoadError
+  nil
+end
 
 class Faulty
   module Patch

lib/faulty/storage/redis.rb

@@ -293,7 +293,7 @@ def key(*parts)
       end
 
       def ckey(circuit_name, *parts)
-        key('circuit', circuit_name, *parts)
+        key('circuit', "{#{circuit_name}}", *parts)
       end
 
       # @return [String] The key for circuit options
@@ -323,7 +323,7 @@ def opened_at_key(circuit_name)
 
       # Get the current key to add circuit names to
       def list_key
-        key('list', current_list_block)
+        key('list', "{#{current_list_block}}")
       end
 
       # Get all active circuit list keys
@@ -348,7 +348,7 @@ def all_list_keys
         num_blocks = (options.circuit_ttl.to_f / options.list_granularity).floor + 1
         start_block = current_list_block - num_blocks + 1
         num_blocks.times.map do |i|
-          key('list', start_block + i)
+          key('list', "{#{start_block + i}}")
         end
       end
 
@@ -372,11 +372,11 @@ def current_list_block
       #   inside the block
       def watch_exec(key, old, &block)
         redis do |r|
-          r.watch(key) do
-            if old.include?(r.get(key))
-              r.multi(&block)
+          r.watch(key) do |c|
+            if old.include?(c.get(key))
+              c.multi(&block)
             else
-              r.unwatch
+              c.unwatch
               nil
             end
           end
@@ -424,11 +424,17 @@ def check_client_options!
         warn "Faulty error while checking client options: #{e.message}"
       end
 
-      def check_redis_options!
+      def check_redis_options! # rubocop:disable Metrics/MethodLength
         gte5 = ::Redis::VERSION.to_f >= 5
-        method = gte5 ? :config : :options
         ropts = redis do |r|
-          r.instance_variable_get(:@client).public_send(method)
+          if r.instance_of?(::Redis)
+            method = gte5 ? :config : :options
+            r._client.public_send(method)
+          elsif r.instance_of?(::Redis::Cluster)
+            r._client.config
+          else
+            raise TypeError, "Unsupported Redis client type: #{r.class}"
+          end
         end
 
         bad_timeouts = {}
@@ -445,7 +451,16 @@ def check_redis_options!
           MSG
         end
 
-        gt1_retry = gte5 ? ropts.retry_connecting?(1, nil) : ropts[:reconnect_attempts] > 1
+        gt1_retry = redis do |r|
+          if r.instance_of?(::Redis)
+            gte5 ? ropts.retry_connecting?(1, nil) : ropts[:reconnect_attempts] > 1
+          elsif r.instance_of?(::Redis::Cluster)
+            ra = ropts.client_config[:reconnect_attempts]
+            (ra.is_a?(Array) && ra.length > 1) || ra > 1
+          else
+            raise TypeError, "Unsupported Redis client type: #{r.class}"
+          end
+        end
         if gt1_retry
           warn <<~MSG
             Faulty recommends setting Redis reconnect_attempts to <= 1 to

spec/circuit_spec.rb

@@ -331,7 +331,13 @@
   end
 
   context 'with redis storage' do
-    let(:storage) { Faulty::Storage::Redis.new }
+    let(:storage) do
+      if ENV['REDIS_CLUSTER'] == 'true'
+        Faulty::Storage::Redis.new(client: Redis::Cluster.new(timeout: 1))
+      else
+        Faulty::Storage::Redis.new
+      end
+    end
 
     after { circuit.reset! }
 
@@ -341,7 +347,11 @@
   context 'with fault-tolerant redis storage' do
     let(:storage) do
       Faulty::Storage::FaultTolerantProxy.new(
-        Faulty::Storage::Redis.new,
+        if ENV['REDIS_CLUSTER'] == 'true'
+          Faulty::Storage::Redis.new(client: Redis::Cluster.new(timeout: 1))
+        else
+          Faulty::Storage::Redis.new
+        end,
         notifier: Faulty::Events::Notifier.new
       )
     end

spec/storage/redis_spec.rb

@@ -1,18 +1,27 @@
 # frozen_string_literal: true
 
 require 'connection_pool'
-require 'redis'
 
 RSpec.describe Faulty::Storage::Redis do
   subject(:storage) { described_class.new(**options.merge(client: client)) }
 
   let(:options) { {} }
-  let(:client) { Redis.new(timeout: 1) }
+  let(:client_options) { { timeout: 1 } }
+  let(:client_class) do
+    if ENV['REDIS_CLUSTER'] == 'true'
+      require 'redis-clustering'
+      Redis::Cluster
+    else
+      require 'redis'
+      Redis
+    end
+  end
+  let(:client) { client_class.new(**client_options) }
   let(:circuit) { Faulty::Circuit.new('test', storage: storage) }
 
   after { circuit&.reset! }
 
-  context 'with default options' do
+  context 'with default options', unless: ENV['REDIS_CLUSTER'] == 'true' do
     subject(:storage) { described_class.new }
 
     it 'can add an entry' do
@@ -32,7 +41,7 @@
     let(:pool_size) { 100 }
 
     let(:client) do
-      ConnectionPool.new(size: pool_size, timeout: 1) { Redis.new(timeout: 1) }
+      ConnectionPool.new(size: pool_size, timeout: 1) { client_class.new(**client_options) }
     end
 
     it 'adds an entry' do
@@ -54,7 +63,7 @@
   end
 
   context 'when Redis has high timeout' do
-    let(:client) { Redis.new(timeout: 5.0) }
+    let(:client) { client_class.new(**client_options, timeout: 5.0) }
 
     it 'prints timeout warning' do
       timeouts = { connect_timeout: 5.0, read_timeout: 5.0, write_timeout: 5.0 }
@@ -63,7 +72,7 @@
   end
 
   context 'when Redis has high reconnect_attempts' do
-    let(:client) { Redis.new(timeout: 1, reconnect_attempts: 2) }
+    let(:client) { client_class.new(**client_options, reconnect_attempts: 2) }
 
     it 'prints reconnect_attempts warning' do
       expect { storage }.to output(/Your setting is larger/).to_stderr
@@ -72,7 +81,7 @@
 
   context 'when ConnectionPool has high timeout' do
     let(:client) do
-      ConnectionPool.new(timeout: 6) { Redis.new(timeout: 1) }
+      ConnectionPool.new(timeout: 6) { client_class.new(**client_options) }
     end
 
     it 'prints timeout warning' do
@@ -82,7 +91,7 @@
 
   context 'when ConnectionPool Redis client has high timeout' do
     let(:client) do
-      ConnectionPool.new(timeout: 1) { Redis.new(timeout: 7.0) }
+      ConnectionPool.new(timeout: 1) { client_class.new(**client_options, timeout: 7.0) }
     end
 
     it 'prints Redis timeout warning' do
@@ -106,16 +115,16 @@
     it 'sets opened_at to the maximum' do
       Timecop.freeze
       storage.open(circuit, Faulty.current_time)
-      client.del('faulty:circuit:test:opened_at')
+      client.del('faulty:circuit:{test}:opened_at')
       status = storage.status(circuit)
       expect(status.opened_at).to eq(Faulty.current_time - storage.options.circuit_ttl)
     end
   end
 
   context 'when history entries are integers and floats' do
     it 'gets floats' do
-      client.lpush('faulty:circuit:test:entries', '1660865630:1')
-      client.lpush('faulty:circuit:test:entries', '1660865646.897674:1')
+      client.lpush('faulty:circuit:{test}:entries', '1660865630:1')
+      client.lpush('faulty:circuit:{test}:entries', '1660865646.897674:1')
       expect(storage.history(circuit)).to eq([[1_660_865_630.0, true], [1_660_865_646.897674, true]])
     end
   end