+
+ def load_cluster_info_from_node(nodeaddr)
+ node = ClusterNode.new(ARGV[1])
+ node.connect(:abort => true)
+ node.assert_cluster
+ node.load_info(:getfriends => true)
+ add_node(node)
+ node.friends.each{|f|
+ fnode = ClusterNode.new(f[:addr])
+ fnode.connect()
+ fnode.load_info()
+ add_node(fnode)
+ }
+ end
+
+ # Given a list of source nodes return a "resharding plan"
+ # with what slots to move in order to move "numslots" slots to another
+ # instance.
+ def compute_reshard_table(sources,numslots)
+ moved = []
+ # 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/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}
+ end
+ }
+ }
+ return moved
+ end
+
+ def show_reshard_table(table)
+ table.each{|e|
+ 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,1000)
+ 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
+
+ # redis-trib subcommands implementations
+
+ def check_cluster_cmd
+ load_cluster_info_from_node(ARGV[1])
+ check_cluster
+ end
+
+ def reshard_cluster_cmd
+ load_cluster_info_from_node(ARGV[1])
+ errors = check_cluster
+ if errors.length != 0
+ puts "Please fix your cluster problems before resharding."
+ exit 1
+ end
+ numslots = 0
+ while numslots <= 0 or numslots > 4096
+ print "How many slots do you want to move (from 1 to 4096)? "
+ numslots = STDIN.gets.to_i
+ end
+ target = nil
+ while not target
+ print "What is the receiving node ID? "
+ target = get_node_by_name(STDIN.gets.chop)
+ if not target
+ puts "The specified node is not known, please retry."
+ end
+ end
+ sources = []
+ puts "Please enter all the source node IDs."
+ puts " Type 'all' to use all the nodes as source nodes for the hash slots."
+ puts " Type 'done' once you entered all the source nodes IDs."
+ while true
+ print "Source node ##{sources.length+1}:"
+ line = STDIN.gets.chop
+ src = get_node_by_name(line)
+ if line == "done"
+ if sources.length == 0
+ puts "No source nodes given, operation aborted"
+ exit 1
+ else
+ break
+ end
+ elsif line == "all"
+ @nodes.each{|n|
+ next if n.info[:name] == target.info[:name]
+ sources << n
+ }
+ break
+ elsif not src
+ puts "The specified node is not known, please retry."
+ elsif src.info[:name] == target.info[:name]
+ puts "It is not possible to use the target node as source node."
+ else
+ sources << src
+ end
+ end
+ puts "\nReady to move #{numslots} slots."
+ puts " Source nodes:"
+ sources.each{|s| puts " "+s.info_string}
+ 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
+ puts "Creating cluster"
+ ARGV[1..-1].each{|n|
+ node = ClusterNode.new(n)
+ node.connect(:abort => true)
+ node.assert_cluster
+ node.load_info
+ node.assert_empty
+ add_node(node)
+ }
+ puts "Performing hash slots allocation on #{@nodes.length} nodes..."
+ alloc_slots
+ show_nodes
+ yes_or_die "Can I set the above configuration?"
+ flush_nodes_config
+ puts "** Nodes configuration updated"
+ puts "** Sending CLUSTER MEET messages to join the cluster"
+ join_cluster
+ check_cluster
+ end
projects / redis.git / blobdiff