fix: the replica affinity option is not working (#109)

Author: supercaracal

.github/workflows/test.yaml

@@ -209,13 +209,13 @@ jobs:
         with:
           ruby-version: '3.1'
           bundler-cache: true
-      - name: Pull Docker images
-        run: docker pull redis:$REDIS_VERSION
       - name: Get IP address of host
         id: host_ip_addr
         run: |
           host_ip_addr=$(ip a | grep eth0 | grep inet | awk '{print $2}' | cut -d'/' -f1)
           echo "::set-output name=host_ip_addr::$host_ip_addr"
+      - name: Pull Docker images
+        run: docker pull redis:$REDIS_VERSION
       - name: Run containers
         run: docker compose -f $DOCKER_COMPOSE_FILE up -d
         env:
@@ -266,8 +266,10 @@ jobs:
     runs-on: ubuntu-latest
     env:
       REDIS_VERSION: '7.0.1'
-      DOCKER_COMPOSE_FILE: 'docker-compose.yaml'
-      REDIS_CLIENT_MAX_THREADS: '37'
+      DOCKER_COMPOSE_FILE: 'docker-compose.latency.yaml'
+      REDIS_REPLICA_SIZE: '2'
+      REDIS_CLIENT_MAX_THREADS: '5'
+      DELAY_TIME: '2ms'
     steps:
       - name: Check out code
         uses: actions/checkout@v3
@@ -284,7 +286,31 @@ jobs:
         run: bundle exec rake wait
       - name: Print containers
         run: docker compose -f $DOCKER_COMPOSE_FILE ps
+      - name: Rebuild cluster for balancing of replicas
+        run: bundle exec rake build_cluster_for_bench
+        env:
+          DEBUG: "1"
+      - name: Print topology
+        run: |
+          for i in {1..9}
+          do
+            echo "node$i: $(docker compose -f $DOCKER_COMPOSE_FILE exec node$i redis-cli cluster nodes | grep myself)"
+          done
+      - name: Ping nodes
+        run: |
+          for i in {1..9}
+          do
+            node_addr="$(docker compose -f $DOCKER_COMPOSE_FILE exec node$i redis-cli cluster nodes | grep myself | awk '{print $2}' | cut -d'@' -f1 | cut -d':' -f1)"
+            echo "node$i:"
+            ping -c 5 $node_addr
+          done
       - name: Run minitest
         run: bundle exec rake bench
+      - name: Reset qdisc
+        run: |
+          for i in {5..9..2}
+          do
+            docker compose -f $DOCKER_COMPOSE_FILE exec node$i tc qdisc del dev eth0 root netem
+          done
       - name: Stop containers
         run: docker compose -f $DOCKER_COMPOSE_FILE down

Rakefile

@@ -62,3 +62,11 @@ task :build_cluster, %i[addr1 addr2] do |_, args|
     timeout: 30.0
   ).rebuild
 end
+
+desc 'Build cluster for benchmark'
+task :build_cluster_for_bench do
+  $LOAD_PATH.unshift(File.expand_path('test', __dir__))
+  require 'cluster_controller'
+  nodes = (6379..6387).map { |port| "redis://127.0.0.1:#{port}" }
+  ::ClusterController.new(nodes, shard_size: 3, replica_size: 2, timeout: 30.0).rebuild
+end

docker-compose.latency.yaml

@@ -0,0 +1,103 @@
+---
+# 3 shards plus with each 2 replicas simutlated network delay
+services:
+  node1: &node
+    image: "redis:${REDIS_VERSION:-7}"
+    command: >
+      redis-server
+      --maxmemory            64mb
+      --maxmemory-policy     allkeys-lru
+      --appendonly           yes
+      --cluster-enabled      yes
+      --cluster-config-file  nodes.conf
+      --cluster-node-timeout 5000
+    restart: "${RESTART_POLICY:-always}"
+    healthcheck:
+      test: ["CMD", "redis-cli", "ping"]
+      interval: "7s"
+      timeout: "5s"
+      retries: 10
+    ports:
+      - "6379:6379"
+  node2:
+    <<: *node
+    ports:
+      - "6380:6379"
+  node3:
+    <<: *node
+    ports:
+      - "6381:6379"
+  node4:
+    <<: *node
+    ports:
+      - "6382:6379"
+  node5: &far_node
+    <<: *node
+    command: >
+      bash -c "apt-get update > /dev/null
+      && apt-get install --no-install-recommends --no-install-suggests -y iproute2 iputils-ping > /dev/null
+      && rm -rf /var/lib/apt/lists/*
+      && tc qdisc add dev eth0 root netem delay ${DELAY_TIME:-20ms}
+      && redis-server
+      --maxmemory            64mb
+      --maxmemory-policy     allkeys-lru
+      --appendonly           yes
+      --cluster-enabled      yes
+      --cluster-config-file  nodes.conf
+      --cluster-node-timeout 5000"
+    cap_add:
+      - NET_ADMIN
+    ports:
+      - "6383:6379"
+  node6:
+    <<: *node
+    ports:
+      - "6384:6379"
+  node7:
+    <<: *far_node
+    ports:
+      - "6385:6379"
+  node8:
+    <<: *node
+    ports:
+      - "6386:6379"
+  node9:
+    <<: *far_node
+    ports:
+      - "6387:6379"
+  clustering:
+    image: "redis:${REDIS_VERSION:-7}"
+    command: >
+      bash -c "apt-get update > /dev/null
+      && apt-get install --no-install-recommends --no-install-suggests -y dnsutils > /dev/null
+      && rm -rf /var/lib/apt/lists/*
+      && yes yes | redis-cli --cluster create
+      $$(dig node1 +short):6379
+      $$(dig node2 +short):6379
+      $$(dig node3 +short):6379
+      $$(dig node4 +short):6379
+      $$(dig node5 +short):6379
+      $$(dig node6 +short):6379
+      $$(dig node7 +short):6379
+      $$(dig node8 +short):6379
+      $$(dig node9 +short):6379
+      --cluster-replicas 2"
+    depends_on:
+      node1:
+        condition: service_healthy
+      node2:
+        condition: service_healthy
+      node3:
+        condition: service_healthy
+      node4:
+        condition: service_healthy
+      node5:
+        condition: service_healthy
+      node6:
+        condition: service_healthy
+      node7:
+        condition: service_healthy
+      node8:
+        condition: service_healthy
+      node9:
+        condition: service_healthy

lib/redis_client/cluster_config.rb

@@ -42,6 +42,17 @@ def initialize( # rubocop:disable Metrics/ParameterLists
       @mutex = Mutex.new
     end
 
+    def dup
+      self.class.new(
+        nodes: @node_configs,
+        replica: @replica,
+        replica_affinity: @replica_affinity,
+        fixed_hostname: @fixed_hostname,
+        client_implementation: @client_implementation,
+        **@client_config
+      )
+    end
+
     def inspect
       "#<#{self.class.name} #{per_node_key.values}>"
     end
@@ -79,10 +90,6 @@ def add_node(host, port)
       @mutex.synchronize { @node_configs << { host: host, port: port } }
     end
 
-    def dup
-      self.class.new(nodes: @node_configs, replica: @replica, fixed_hostname: @fixed_hostname, **@client_config)
-    end
-
     private
 
     def build_node_configs(addrs)

test/bench_command.rb

@@ -94,7 +94,7 @@ def new_test_client
     end
   end
 
-  class ScaleRead < BenchmarkWrapper
+  class ScaleReadRandom < BenchmarkWrapper
     include Mixin
 
     private
@@ -103,6 +103,24 @@ def new_test_client
       config = ::RedisClient::ClusterConfig.new(
         nodes: TEST_NODE_URIS,
         replica: true,
+        replica_affinity: :random,
+        fixed_hostname: TEST_FIXED_HOSTNAME,
+        **TEST_GENERIC_OPTIONS
+      )
+      ::RedisClient::Cluster.new(config)
+    end
+  end
+
+  class ScaleReadLatency < BenchmarkWrapper
+    include Mixin
+
+    private
+
+    def new_test_client
+      config = ::RedisClient::ClusterConfig.new(
+        nodes: TEST_NODE_URIS,
+        replica: true,
+        replica_affinity: :latency,
         fixed_hostname: TEST_FIXED_HOSTNAME,
         **TEST_GENERIC_OPTIONS
       )

test/redis_client/test_cluster.rb

@@ -38,6 +38,7 @@ def test_call
         assert(@client.call('PING') { |r| r == 'PONG' })
 
         assert_equal(2, @client.call('HSET', 'hash', { foo: 1, bar: 2 }))
+        wait_for_replication
         assert_equal(%w[1 2], @client.call('HMGET', 'hash', %w[foo bar]))
       end
 
@@ -53,6 +54,7 @@ def test_call_once
         assert(@client.call_once('PING') { |r| r == 'PONG' })
 
         assert_equal(2, @client.call_once('HSET', 'hash', { foo: 1, bar: 2 }))
+        wait_for_replication
         assert_equal(%w[1 2], @client.call_once('HMGET', 'hash', %w[foo bar]))
       end
 

test/redis_client/test_cluster_config.rb

@@ -5,6 +5,23 @@
 
 class RedisClient
   class TestClusterConfig < TestingWrapper
+    def test_dup
+      orig = ::RedisClient::ClusterConfig.new
+      copy = orig.dup
+      refute_equal(orig.object_id, copy.object_id)
+
+      ::RedisClient::ClusterConfig.instance_method(:initialize).parameters.each do |type, name|
+        case type
+        when :key
+          want = orig.instance_variable_get("@#{name}".to_sym)
+          got = copy.instance_variable_get("@#{name}".to_sym)
+          next if got.nil?
+
+          assert_equal(want, got, "Case: #{type}=#{name}")
+        end
+      end
+    end
+
     def test_inspect
       want = '#<RedisClient::ClusterConfig [{:host=>"127.0.0.1", :port=>6379}]>'
       got = ::RedisClient::ClusterConfig.new.inspect
@@ -100,12 +117,6 @@ def test_add_node
       assert_equal([{ host: '127.0.0.1', port: 6379 }, { host: '127.0.0.2', port: 6380 }], config.instance_variable_get(:@node_configs))
     end
 
-    def test_dup
-      orig = ::RedisClient::ClusterConfig.new
-      copy = orig.dup
-      refute_equal(orig.object_id, copy.object_id)
-    end
-
     def test_command_builder
       assert_equal(::RedisClient::CommandBuilder, ::RedisClient::ClusterConfig.new.command_builder)
     end