Class: RedisClient::Cluster::Router

Inherits:
Object
  • Object
show all
Defined in:
lib/redis_client/cluster/router.rb

Instance Method Summary collapse

Constructor Details

#initialize(config, concurrent_worker, pool: nil, **kwargs) ⇒ Router 



24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/redis_client/cluster/router.rb', line 24

def initialize(config, concurrent_worker, pool: nil, **kwargs)
  @config = config.dup
  @original_config = config.dup if config.connect_with_original_config
  @connect_with_original_config = config.connect_with_original_config
  @concurrent_worker = concurrent_worker
  @pool = pool
  @client_kwargs = kwargs
  @node = ::RedisClient::Cluster::Node.new(concurrent_worker, config: config, pool: pool, **kwargs)
  @node.reload!
  @command = ::RedisClient::Cluster::Command.load(@node.replica_clients.shuffle, slow_command_timeout: config.slow_command_timeout)
  @command_builder = @config.command_builder
end

Instance Method Details

#assign_asking_node(err_msg) ⇒ Object 



233
234
235
236
 
# File 'lib/redis_client/cluster/router.rb', line 233

def assign_asking_node(err_msg)
  _, _, node_key = err_msg.split
  handle_node_reload_error { @node.find_by(node_key) }
end

#assign_node(command) ⇒ Object 



168
169
170
171
172
173
 
# File 'lib/redis_client/cluster/router.rb', line 168

def assign_node(command)
  handle_node_reload_error do
    node_key = find_node_key(command)
    @node.find_by(node_key)
  end
end

#assign_redirection_node(err_msg) ⇒ Object 



226
227
228
229
230
231
 
# File 'lib/redis_client/cluster/router.rb', line 226

def assign_redirection_node(err_msg)
  _, slot, node_key = err_msg.split
  slot = slot.to_i
  @node.update_slot(slot, node_key)
  handle_node_reload_error { @node.find_by(node_key) }
end

#close ⇒ Object 



248
249
250
 
# File 'lib/redis_client/cluster/router.rb', line 248

def close
  @node.each(&:close)
end

#command_exists?(name) ⇒ Boolean 



222
223
224
 
# File 'lib/redis_client/cluster/router.rb', line 222

def command_exists?(name)
  @command.exists?(name)
end

#find_node(node_key) ⇒ Object 



218
219
220
 
# File 'lib/redis_client/cluster/router.rb', line 218

def find_node(node_key)
  handle_node_reload_error { @node.find_by(node_key) }
end

#find_node_key(command, seed: nil) ⇒ Object 



196
197
198
199
 
# File 'lib/redis_client/cluster/router.rb', line 196

def find_node_key(command, seed: nil)
  key = @command.extract_first_key(command)
  find_node_key_by_key(key, seed: seed, primary: @command.should_send_to_primary?(command))
end

#find_node_key_by_key(key, seed: nil, primary: false) ⇒ Object 



175
176
177
178
179
180
181
182
183
184
185
186
187
 
# File 'lib/redis_client/cluster/router.rb', line 175

def find_node_key_by_key(key, seed: nil, primary: false)
  if key && !key.empty?
    slot = ::RedisClient::Cluster::KeySlotConverter.convert(key)
    node_key = primary ? @node.find_node_key_of_primary(slot) : @node.find_node_key_of_replica(slot)
    if node_key.nil?
      renew_cluster_state
      raise ::RedisClient::Cluster::NodeMightBeDown
    end
    node_key
  else
    primary ? @node.any_primary_node_key(seed: seed) : @node.any_replica_node_key(seed: seed)
  end
end

#find_primary_node_by_slot(slot) ⇒ Object 



189
190
191
192
193
194
 
# File 'lib/redis_client/cluster/router.rb', line 189

def find_primary_node_by_slot(slot)
  handle_node_reload_error do
    node_key = @node.find_node_key_of_primary(slot)
    @node.find_by(node_key)
  end
end

#find_primary_node_key(command) ⇒ Object 



201
202
203
204
205
206
 
# File 'lib/redis_client/cluster/router.rb', line 201

def find_primary_node_key(command)
  key = @command.extract_first_key(command)
  return nil unless key&.size&.> 0

  find_node_key_by_key(key, primary: true)
end

#find_slot(command) ⇒ Object 



208
209
210
 
# File 'lib/redis_client/cluster/router.rb', line 208

def find_slot(command)
  find_slot_by_key(@command.extract_first_key(command))
end

#find_slot_by_key(key) ⇒ Object 



212
213
214
215
216
 
# File 'lib/redis_client/cluster/router.rb', line 212

def find_slot_by_key(key)
  return if key.empty?

  ::RedisClient::Cluster::KeySlotConverter.convert(key)
end

#handle_redirection(node, retry_count:) ⇒ Object

rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity



109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/redis_client/cluster/router.rb', line 109

def handle_redirection(node, retry_count:) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
  yield node
rescue ::RedisClient::CircuitBreaker::OpenCircuitError
  raise
rescue ::RedisClient::CommandError => e
  raise unless ::RedisClient::Cluster::ErrorIdentification.client_owns_error?(e, node)

  retry_count -= 1
  if e.message.start_with?('MOVED')
    node = assign_redirection_node(e.message)
    retry if retry_count >= 0
  elsif e.message.start_with?('ASK')
    node = assign_asking_node(e.message)
    if retry_count >= 0
      node.call('ASKING')
      retry
    end
  elsif e.message.start_with?('CLUSTERDOWN Hash slot not served')
    renew_cluster_state
    retry if retry_count >= 0
  end

  raise
rescue ::RedisClient::ConnectionError => e
  raise unless ::RedisClient::Cluster::ErrorIdentification.client_owns_error?(e, node)

  retry_count -= 1
  renew_cluster_state
  retry if retry_count >= 0
  raise
end

#node_keys ⇒ Object 



238
239
240
 
# File 'lib/redis_client/cluster/router.rb', line 238

def node_keys
  @node.node_keys
end

#renew_cluster_state ⇒ Object 



242
243
244
245
246
 
# File 'lib/redis_client/cluster/router.rb', line 242

def renew_cluster_state
  @node.reload!
rescue ::RedisClient::Cluster::InitialSetupError
  # ignore
end

#scan(*command, seed: nil, **kwargs) ⇒ Object

rubocop:disable Metrics/AbcSize



141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
 
# File 'lib/redis_client/cluster/router.rb', line 141

def scan(*command, seed: nil, **kwargs) # rubocop:disable Metrics/AbcSize
  command = @command_builder.generate(command, kwargs)

  command[1] = ZERO_CURSOR_FOR_SCAN if command.size == 1
  input_cursor = Integer(command[1])

  client_index = input_cursor % 256
  raw_cursor = input_cursor >> 8

  clients = @node.clients_for_scanning(seed: seed)

  client = clients[client_index]
  return [ZERO_CURSOR_FOR_SCAN, []] unless client

  command[1] = raw_cursor.to_s

  result_cursor, result_keys = client.call_v(command)
  result_cursor = Integer(result_cursor)

  client_index += 1 if result_cursor == 0

  [((result_cursor << 8) + client_index).to_s, result_keys]
rescue ::RedisClient::ConnectionError
  renew_cluster_state
  raise
end

#send_command(method, command, *args, &block) ⇒ Object

rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/redis_client/cluster/router.rb', line 37

def send_command(method, command, *args, &block) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
  cmd = ::RedisClient::Cluster::NormalizedCmdName.instance.get_by_command(command)
  case cmd
  when 'ping'     then @node.send_ping(method, command, args).first.then(&TSF.call(block))
  when 'wait'     then send_wait_command(method, command, args, &block)
  when 'keys'     then @node.call_replicas(method, command, args).flatten.sort_by(&:to_s).then(&TSF.call(block))
  when 'dbsize'   then @node.call_replicas(method, command, args).select { |e| e.is_a?(Integer) }.sum.then(&TSF.call(block))
  when 'scan'     then scan(command, seed: 1)
  when 'lastsave' then @node.call_all(method, command, args).sort_by(&:to_i).then(&TSF.call(block))
  when 'role'     then @node.call_all(method, command, args, &block)
  when 'config'   then send_config_command(method, command, args, &block)
  when 'client'   then send_client_command(method, command, args, &block)
  when 'cluster'  then send_cluster_command(method, command, args, &block)
  when 'memory'   then send_memory_command(method, command, args, &block)
  when 'script'   then send_script_command(method, command, args, &block)
  when 'pubsub'   then send_pubsub_command(method, command, args, &block)
  when 'watch'    then send_watch_command(command, &block)
  when 'mset', 'mget', 'del'
    send_multiple_keys_command(cmd, method, command, args, &block)
  when 'acl', 'auth', 'bgrewriteaof', 'bgsave', 'quit', 'save'
    @node.call_all(method, command, args).first.then(&TSF.call(block))
  when 'flushall', 'flushdb'
    @node.call_primaries(method, command, args).first.then(&TSF.call(block))
  when 'readonly', 'readwrite', 'shutdown'
    raise ::RedisClient::Cluster::OrchestrationCommandNotSupported, cmd
  when 'discard', 'exec', 'multi', 'unwatch'
    raise ::RedisClient::Cluster::AmbiguousNodeError, cmd
  else
    node = assign_node(command)
    try_send(node, method, command, args, &block)
  end
rescue ::RedisClient::CircuitBreaker::OpenCircuitError
  raise
rescue ::RedisClient::Cluster::Node::ReloadNeeded
  renew_cluster_state
  raise ::RedisClient::Cluster::NodeMightBeDown
rescue ::RedisClient::ConnectionError
  renew_cluster_state
  raise
rescue ::RedisClient::CommandError => e
  renew_cluster_state if e.message.start_with?('CLUSTERDOWN Hash slot not served')
  raise
rescue ::RedisClient::Cluster::ErrorCollection => e
  raise if e.errors.any?(::RedisClient::CircuitBreaker::OpenCircuitError)

  renew_cluster_state if e.errors.values.any? do |err|
    next false if ::RedisClient::Cluster::ErrorIdentification.identifiable?(err) && @node.none? { |c| ::RedisClient::Cluster::ErrorIdentification.client_owns_error?(err, c) }

    err.message.start_with?('CLUSTERDOWN Hash slot not served') || err.is_a?(::RedisClient::ConnectionError)
  end

  raise
end

#try_delegate(node, method, *args, retry_count: 3, **kwargs, &block) ⇒ Object 



103
104
105
106
107
 
# File 'lib/redis_client/cluster/router.rb', line 103

def try_delegate(node, method, *args, retry_count: 3, **kwargs, &block)
  handle_redirection(node, retry_count: retry_count) do |on_node|
    on_node.public_send(method, *args, **kwargs, &block)
  end
end

#try_send(node, method, command, args, retry_count: 3, &block) ⇒ Object

See Also:



92
93
94
95
96
97
98
99
100
101
 
# File 'lib/redis_client/cluster/router.rb', line 92

def try_send(node, method, command, args, retry_count: 3, &block)
  handle_redirection(node, retry_count: retry_count) do |on_node|
    if args.empty?
      # prevent memory allocation for variable-length args
      on_node.public_send(method, command, &block)
    else
      on_node.public_send(method, *args, command, &block)
    end
  end
end