#!/usr/bin/env ruby
+# TODO (temporary here, we'll move this into the Github issues once
+# redis-trib initial implementation is complted).
+#
+# - Make sure that if the rehashing fails in the middle redis-trib will try
+# to recover.
+# - When redis-trib performs a cluster check, if it detects a slot move in
+# progress it should prompt the user to continue the move from where it
+# stopped.
+# - Gracefully handle Ctrl+C in move_slot to prompt the user if really stop
+# while rehashing, and performing the best cleanup possible if the user
+# forces the quit.
+# - When doing "fix" set a global Fix to true, and prompt the user to
+# fix the problem if automatically fixable every time there is something
+# to fix. For instance:
+# 1) If there is a node that pretend to receive a slot, or to migrate a
+# slot, but has no entries in that slot, fix it.
+# 2) If there is a node having keys in slots that are not owned by it
+# fix this condiiton moving the entries in the same node.
+# 3) Perform more possibly slow tests about the state of the cluster.
+# 4) When aborted slot migration is detected, fix it.
+
require 'rubygems'
require 'redis'
nodes = @r.cluster("nodes").split("\n")
nodes.each{|n|
# name addr flags role ping_sent ping_recv link_status slots
- name,addr,flags,role,ping_sent,ping_recv,link_status,slots = n.split(" ")
+ split = n.split
+ name,addr,flags,role,ping_sent,ping_recv,link_status = split[0..6]
+ slots = split[7..-1]
info = {
:name => name,
:addr => addr,
if info[:flags].index("myself")
@info = @info.merge(info)
@info[:slots] = {}
- slots.split(",").each{|s|
- if s.index("-")
+ slots.each{|s|
+ if s[0..0] == '['
+ # Fixme: for now skipping migration entries
+ elsif s.index("-")
start,stop = s.split("-")
self.add_slots((start.to_i)..(stop.to_i))
else
x.count == 1 ? x.first.to_s : "#{x.first}-#{x.last}"
}.join(",")
- "[#{@info[:cluster_state].upcase}] #{self.info[:name]} #{self.to_s.ljust(25)} slots:#{slots}"
+ "[#{@info[:cluster_state].upcase}] #{self.info[:name]} #{self.to_s} slots:#{slots} (#{self.slots.length} slots)"
end
def info
# instance.
def compute_reshard_table(sources,numslots)
moved = []
- sources.each{|s|
+ # Sort from bigger to smaller instance, for two reasons:
+ # 1) If we take less slots than instanes it is better to start getting from
+ # the biggest instances.
+ # 2) We take one slot more from the first instance in the case of not perfect
+ # divisibility. Like we have 3 nodes and need to get 10 slots, we take
+ # 4 from the first, and 3 from the rest. So the biggest is always the first.
+ sources = sources.sort{|a,b| b.slots.length <=> a.slots.length}
+ source_tot_slots = sources.inject(0) {|sum,source| sum+source.slots.length}
+ sources.each_with_index{|s,i|
# Every node will provide a number of slots proportional to the
# slots it has assigned.
- n = (numslots.to_f/4096*s.slots.length).ceil
+ n = (numslots.to_f/source_tot_slots*s.slots.length)
+ if i == 0
+ n = n.ceil
+ else
+ n = n.floor
+ end
s.slots.keys.sort[(0...n)].each{|slot|
if moved.length < numslots
moved << {:source => s, :slot => slot}
def show_reshard_table(table)
table.each{|e|
- puts "Moving slot #{e[:slot]} from #{e[:source].info[:name]}"
+ puts " Moving slot #{e[:slot]} from #{e[:source].info[:name]}"
+ }
+ end
+
+ def move_slot(source,target,slot,o={})
+ # We start marking the slot as importing in the destination node,
+ # and the slot as migrating in the target host. Note that the order of
+ # the operations is important, as otherwise a client may be redirected to
+ # the target node that does not yet know it is importing this slot.
+ print "Moving slot #{slot} from #{source.info_string}: "; STDOUT.flush
+ target.r.cluster("setslot",slot,"importing",source.info[:name])
+ source.r.cluster("setslot",slot,"migrating",source.info[:name])
+ # Migrate all the keys from source to target using the MIGRATE command
+ while true
+ keys = source.r.cluster("getkeysinslot",slot,10)
+ break if keys.length == 0
+ keys.each{|key|
+ source.r.migrate(target.info[:host],target.info[:port],key,0,1)
+ print "." if o[:verbose]
+ STDOUT.flush
+ }
+ end
+ puts
+ # Set the new node as the owner of the slot in all the known nodes.
+ @nodes.each{|n|
+ n.r.cluster("setslot",slot,"node",target.info[:name])
}
end
end
numslots = 0
while numslots <= 0 or numslots > 4096
- print "How many slots do you want to move (from 1 to 4096)?"
+ print "How many slots do you want to move (from 1 to 4096)? "
numslots = STDIN.gets.to_i
end
target = nil
puts " Destination node:"
puts " #{target.info_string}"
reshard_table = compute_reshard_table(sources,numslots)
+ puts " Resharding plan:"
show_reshard_table(reshard_table)
+ print "Do you want to proceed with the proposed reshard plan (yes/no)? "
+ yesno = STDIN.gets.chop
+ exit(1) if (yesno != "yes")
+ reshard_table.each{|e|
+ move_slot(e[:source],target,e[:slot],:verbose=>true)
+ }
end
def create_cluster_cmd
node = ClusterNode.new(n)
node.connect(:abort => true)
node.assert_cluster
+ node.load_info
node.assert_empty
add_node(node)
}
projects / redis.git / blobdiff