Add some test cases (#11)

Author: supercaracal

.github/workflows/test.yaml

@@ -51,6 +51,8 @@ jobs:
       - name: Docker compose up
         run: docker compose up -d
       - name: Wait for Redis cluster to be ready
+        run: bundle exec rake wait
+      - name: Print containers
         run: docker compose ps
       - name: Run minitest
         run: bundle exec rake test

Rakefile

@@ -4,9 +4,17 @@ require 'rake/testtask'
 
 task default: :test
 
-desc 'execute all test'
 Rake::TestTask.new :test do |t|
   t.libs << :test
   t.libs << :lib
   t.test_files = Dir['test/**/test_*.rb']
 end
+
+desc 'Wait for cluster to be ready'
+task :wait do
+  $LOAD_PATH.unshift(File.expand_path('test', __dir__))
+  require 'redis_client/cluster/controller'
+  nodes = (7000..7005).map { |port| "redis://127.0.0.1:#{port}" }
+  ctrl = ::RedisClient::Cluster::Controller.new(nodes)
+  ctrl.wait_for_cluster_to_be_ready
+end

lib/redis_client/cluster/command.rb

@@ -9,7 +9,8 @@ class Command
       class << self
         def load(nodes)
           errors = nodes.map do |node|
-            details = fetch_command_details(node)
+            reply = node.call('COMMAND')
+            details = parse_command_details(reply)
             return ::RedisClient::Cluster::Command.new(details)
           rescue ::RedisClient::ConnectionError, ::RedisClient::CommandError => e
             e
@@ -20,9 +21,9 @@ def load(nodes)
 
         private
 
-        def fetch_command_details(node)
-          node.call('COMMAND').to_h do |reply|
-            [reply[0], { arity: reply[1], flags: reply[2], first: reply[3], last: reply[4], step: reply[5] }]
+        def parse_command_details(rows)
+          rows.to_h do |row|
+            [row[0], { arity: row[1], flags: row[2], first: row[3], last: row[4], step: row[5] }]
           end
         end
       end
@@ -51,7 +52,7 @@ def should_send_to_replica?(command)
       private
 
       def pick_details(details)
-        details.transform_values do |detail|
+        (details || {}).transform_values do |detail|
           {
             first_key_position: detail[:first],
             write: detail[:flags].include?('write'),

test/redis_client/cluster/controller.rb

@@ -0,0 +1,292 @@
+# frozen_string_literal: true
+
+require 'redis_client'
+
+class RedisClient
+  class Cluster
+    class Controller
+      SLOT_SIZE = 16_384
+
+      def initialize(node_addrs, timeout: 30.0, reconnect_attempts: 10)
+        raise 'Redis Cluster requires at least 3 master nodes.' if node_addrs.size < 3
+
+        @clients = node_addrs.map do |addr|
+          RedisClient.new(url: addr, timeout: timeout, reconnect_attempts: reconnect_attempts)
+        end
+
+        @timeout = timeout
+      end
+
+      def wait_for_cluster_to_be_ready
+        wait_meeting(@clients)
+        wait_cluster_building(@clients)
+        wait_replication(@clients)
+        wait_cluster_recovering(@clients)
+      end
+
+      def rebuild
+        flush_all_data(@clients)
+        reset_cluster(@clients)
+        assign_slots(@clients)
+        save_config_epoch(@clients)
+        meet_each_other(@clients)
+        wait_meeting(@clients)
+        replicate(@clients)
+        save_config(@clients)
+        wait_cluster_building(@clients)
+        wait_replication(@clients)
+        wait_cluster_recovering(@clients)
+      end
+
+      def down
+        flush_all_data(@clients)
+        reset_cluster(@clients)
+      end
+
+      def fail_serving_master
+        master, slave = take_replication_pairs(@clients)
+        master.shutdown
+        attempt_count = 1
+        max_attempts = 500
+        attempt_count.step(max_attempts) do |i|
+          break if slave.role == 'master' || i >= max_attempts
+
+          attempt_count += 1
+          sleep 0.1
+        end
+      end
+
+      def failover
+        master, slave = take_replication_pairs(@clients)
+        wait_replication_delay(@clients, @timeout)
+        slave.cluster(:failover, :takeover)
+        wait_failover(to_node_key(master), to_node_key(slave), @clients)
+        wait_replication_delay(@clients, @timeout)
+        wait_cluster_recovering(@clients)
+      end
+
+      def start_resharding(slot, src_node_key, dest_node_key, slice_size: 10)
+        node_map = hashify_node_map(@clients.first)
+        src_node_id = node_map.fetch(src_node_key)
+        src_client = find_client(@clients, src_node_key)
+        dest_node_id = node_map.fetch(dest_node_key)
+        dest_client = find_client(@clients, dest_node_key)
+        dest_host, dest_port = dest_node_key.split(':')
+
+        dest_client.call('CLUSTER', 'SETSLOT', slot, 'IMPORTING', src_node_id)
+        src_client.call('CLUSTER', 'SETSLOT', slot, 'MIGRATING', dest_node_id)
+
+        keys_count = src_client.call('CLUSTER', 'COUNTKEYSINSLOT', slot)
+        loop do
+          break if keys_count <= 0
+
+          keys = src_client.call('CLUSTER', 'GETKEYSINSLOT', slot, slice_size)
+          break if keys.empty?
+
+          keys.each do |k|
+            src_client.call('MIGRATE', dest_host, dest_port, k)
+          rescue ::RedisClient::CommandError => e
+            raise unless e.message.start_with?('IOERR')
+
+            src_client.call('MIGRATE', dest_host, dest_port, k, 'REPLACE') # retry once
+          ensure
+            keys_count -= 1
+          end
+        end
+      end
+
+      def finish_resharding(slot, dest_node_key)
+        node_map = hashify_node_map(@clients.first)
+        @clients.first.call('CLUSTER', 'SETSLOT', slot, 'NODE', node_map.fetch(dest_node_key))
+      end
+
+      def close
+        @clients.each(&:close)
+      end
+
+      private
+
+      def flush_all_data(clients)
+        clients.each do |c|
+          c.flushall
+        rescue ::RedisClient::CommandError
+          # READONLY You can't write against a read only slave.
+          nil
+        end
+      end
+
+      def reset_cluster(clients)
+        clients.each { |c| c.cluster(:reset) }
+      end
+
+      def assign_slots(clients)
+        masters = take_masters(clients)
+        slot_slice = SLOT_SIZE / masters.size
+        mod = SLOT_SIZE % masters.size
+        slot_sizes = Array.new(masters.size, slot_slice)
+        mod.downto(1) { |i| slot_sizes[i] += 1 }
+
+        slot_idx = 0
+        masters.zip(slot_sizes).each do |c, s|
+          slot_range = slot_idx..slot_idx + s - 1
+          c.cluster(:addslots, *slot_range.to_a)
+          slot_idx += s
+        end
+      end
+
+      def save_config_epoch(clients)
+        clients.each_with_index do |c, i|
+          c.cluster('set-config-epoch', i + 1)
+        rescue ::RedisClient::CommandError
+          # ERR Node config epoch is already non-zero
+          nil
+        end
+      end
+
+      def meet_each_other(clients)
+        clients.each do |client|
+          next if client.id == clients.first.id
+
+          client.cluster(:meet, target_host, target_port)
+        end
+      end
+
+      def wait_meeting(clients, max_attempts: 600)
+        size = clients.size.to_s
+
+        wait_for_state(clients, max_attempts) do |client|
+          info = hashify_cluster_info(client)
+          info['cluster_known_nodes'] == size
+        end
+      end
+
+      def replicate(clients)
+        node_map = hashify_node_map(clients.first)
+        masters = take_masters(clients)
+
+        take_slaves(clients).each_with_index do |slave, i|
+          master_info = masters[i].connection
+          master_host = master_info.fetch(:host)
+          master_port = master_info.fetch(:port)
+
+          loop do
+            begin
+              master_node_id = node_map.fetch("#{master_host}:#{master_port}")
+              slave.cluster(:replicate, master_node_id)
+            rescue ::RedisClient::CommandError
+              # ERR Unknown node [key]
+              sleep 0.1
+              node_map = hashify_node_map(clients.first)
+              next
+            end
+
+            break
+          end
+        end
+      end
+
+      def save_config(clients)
+        clients.each { |c| c.cluster(:saveconfig) }
+      end
+
+      def wait_cluster_building(clients, max_attempts: 600)
+        wait_for_state(clients, max_attempts) do |client|
+          info = hashify_cluster_info(client)
+          info['cluster_state'] == 'ok'
+        end
+      end
+
+      def wait_replication(clients, max_attempts: 600)
+        wait_for_state(clients, max_attempts) do |client|
+          flags = hashify_cluster_node_flags(client)
+          flags.values.count { |f| f == 'slave' } == 3
+        end
+      end
+
+      def wait_failover(master_key, slave_key, clients, max_attempts: 600)
+        wait_for_state(clients, max_attempts) do |client|
+          flags = hashify_cluster_node_flags(client)
+          flags[master_key] == 'slave' && flags[slave_key] == 'master'
+        end
+      end
+
+      def wait_replication_delay(clients, timeout_sec)
+        timeout_msec = timeout_sec.to_i * 1000
+        wait_for_state(clients, clients.size + 1) do |client|
+          client.blocking_call('WAIT', 1, timeout_msec) if client.call('ROLE').first == 'master'
+          true
+        end
+      end
+
+      def wait_cluster_recovering(clients, max_attempts: 600)
+        key = 0
+        wait_for_state(clients, max_attempts) do |client|
+          client.call('GET', key) if client.call('ROLE').first == 'master'
+          true
+        rescue ::RedisClient::CommandError => e
+          if e.message.start_with?('CLUSTERDOWN')
+            false
+          elsif e.message.start_with?('MOVED')
+            key += 1
+            false
+          else
+            true
+          end
+        end
+      end
+
+      def wait_for_state(clients, max_attempts)
+        attempt_count = 1
+        clients.each do |client|
+          attempt_count.step(max_attempts) do |i|
+            break if i >= max_attempts
+
+            attempt_count += 1
+            break if yield(client)
+
+            sleep 0.1
+          end
+        end
+      end
+
+      def hashify_cluster_info(client)
+        client.call('CLUSTER', 'INFO').split("\r\n").to_h { |str| str.split(':') }
+      end
+
+      def hashify_cluster_node_flags(client)
+        client.call('CLUSTER', 'NODES').split("\n").map(&:split)
+              .to_h { |arr| [arr[1].split('@').first, (arr[2].split(',') & %w[master slave]).first] }
+      end
+
+      def hashify_node_map(client)
+        client.call('CLUSTER', 'NODES').split("\n").map(&:split).to_h { |arr| [arr[1].split('@').first, arr[0]] }
+      end
+
+      def take_masters(clients)
+        size = clients.size / 2
+        return clients if size < 3
+
+        clients.take(size)
+      end
+
+      def take_slaves(clients)
+        size = clients.size / 2
+        return [] if size < 3
+
+        clients[size..size * 2]
+      end
+
+      def take_replication_pairs(clients)
+        [take_masters(clients).last, take_slaves(clients).last]
+      end
+
+      def find_client(clients, node_key)
+        clients.find { |cli| node_key == to_node_key(cli) }
+      end
+
+      def to_node_key(client)
+        "#{client.config.host}:#{client.config.port}"
+      end
+    end
+  end
+end