From 31e2156c4b9a58b88a90c64eecc4bc96cd5de2e4 Mon Sep 17 00:00:00 2001 From: antirez Date: Tue, 3 Apr 2012 10:31:51 +0200 Subject: [PATCH] redis-trib.rb removed from 2.6 branch. --- src/redis-trib.rb | 492 ---------------------------------------------- 1 file changed, 492 deletions(-) delete mode 100755 src/redis-trib.rb diff --git a/src/redis-trib.rb b/src/redis-trib.rb deleted file mode 100755 index 473e4922..00000000 --- a/src/redis-trib.rb +++ /dev/null @@ -1,492 +0,0 @@ -#!/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' - -ClusterHashSlots = 4096 - -def xputs(s) - printf s - STDOUT.flush -end - -class ClusterNode - def initialize(addr) - s = addr.split(":") - if s.length != 2 - puts "Invalid node name #{addr}" - exit 1 - end - @r = nil - @info = {} - @info[:host] = s[0] - @info[:port] = s[1] - @info[:slots] = {} - @dirty = false # True if we need to flush slots info into node. - @friends = [] - end - - def friends - @friends - end - - def slots - @info[:slots] - end - - def to_s - "#{@info[:host]}:#{@info[:port]}" - end - - def connect(o={}) - return if @r - xputs "Connecting to node #{self}: " - begin - @r = Redis.new(:host => @info[:host], :port => @info[:port]) - @r.ping - rescue - puts "ERROR" - puts "Sorry, can't connect to node #{self}" - exit 1 if o[:abort] - @r = nil - end - puts "OK" - end - - def assert_cluster - info = @r.info - if !info["cluster_enabled"] || info["cluster_enabled"].to_i == 0 - puts "Error: Node #{self} is not configured as a cluster node." - exit 1 - end - end - - def assert_empty - if !(@r.cluster("info").split("\r\n").index("cluster_known_nodes:1")) || - (@r.info['db0']) - puts "Error: Node #{self} is not empty. Either the node already knows other nodes (check with nodes-info) or contains some key in database 0." - exit 1 - end - end - - def load_info(o={}) - self.connect - nodes = @r.cluster("nodes").split("\n") - nodes.each{|n| - # name addr flags role ping_sent ping_recv link_status slots - 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, - :flags => flags.split(","), - :role => role, - :ping_sent => ping_sent.to_i, - :ping_recv => ping_recv.to_i, - :link_status => link_status - } - if info[:flags].index("myself") - @info = @info.merge(info) - @info[:slots] = {} - 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 - self.add_slots((s.to_i)..(s.to_i)) - end - } if slots - @dirty = false - @r.cluster("info").split("\n").each{|e| - k,v=e.split(":") - k = k.to_sym - v.chop! - if k != :cluster_state - @info[k] = v.to_i - else - @info[k] = v - end - } - elsif o[:getfriends] - @friends << info - end - } - end - - def add_slots(slots) - slots.each{|s| - @info[:slots][s] = :new - } - @dirty = true - end - - def flush_node_config - return if !@dirty - new = [] - @info[:slots].each{|s,val| - if val == :new - new << s - @info[:slots][s] = true - end - } - @r.cluster("addslots",*new) - @dirty = false - end - - def info_string - # We want to display the hash slots assigned to this node - # as ranges, like in: "1-5,8-9,20-25,30" - # - # Note: this could be easily written without side effects, - # we use 'slots' just to split the computation into steps. - - # First step: we want an increasing array of integers - # for instance: [1,2,3,4,5,8,9,20,21,22,23,24,25,30] - slots = @info[:slots].keys.sort - - # As we want to aggregate adiacent slots we convert all the - # slot integers into ranges (with just one element) - # So we have something like [1..1,2..2, ... and so forth. - slots.map!{|x| x..x} - - # Finally we group ranges with adiacent elements. - slots = slots.reduce([]) {|a,b| - if !a.empty? && b.first == (a[-1].last)+1 - a[0..-2] + [(a[-1].first)..(b.last)] - else - a + [b] - end - } - - # Now our task is easy, we just convert ranges with just one - # element into a number, and a real range into a start-end format. - # Finally we join the array using the comma as separator. - slots = slots.map{|x| - x.count == 1 ? x.first.to_s : "#{x.first}-#{x.last}" - }.join(",") - - "[#{@info[:cluster_state].upcase}] #{self.info[:name]} #{self.to_s} slots:#{slots} (#{self.slots.length} slots)" - end - - def info - @info - end - - def is_dirty? - @dirty - end - - def r - @r - end -end - -class RedisTrib - def initialize - @nodes = [] - end - - def check_arity(req_args, num_args) - if ((req_args > 0 and num_args != req_args) || - (req_args < 0 and num_args < req_args.abs)) - puts "Wrong number of arguments for specified sub command" - exit 1 - end - end - - def add_node(node) - @nodes << node - end - - def get_node_by_name(name) - @nodes.each{|n| - return n if n.info[:name] == name.downcase - } - return nil - end - - def check_cluster - puts "Performing Cluster Check (using node #{@nodes[0]})" - errors = [] - show_nodes - # Check if all the slots are covered - slots = {} - @nodes.each{|n| - slots = slots.merge(n.slots) - } - if slots.length == 4096 - puts "[OK] All 4096 slots covered." - else - errors << "[ERR] Not all 4096 slots are covered by nodes." - puts errors[-1] - end - return errors - end - - def alloc_slots - slots_per_node = ClusterHashSlots/@nodes.length - i = 0 - @nodes.each{|n| - first = i*slots_per_node - last = first+slots_per_node-1 - last = ClusterHashSlots-1 if i == @nodes.length-1 - n.add_slots first..last - i += 1 - } - end - - def flush_nodes_config - @nodes.each{|n| - n.flush_node_config - } - end - - def show_nodes - @nodes.each{|n| - puts n.info_string - } - end - - def join_cluster - # We use a brute force approach to make sure the node will meet - # each other, that is, sending CLUSTER MEET messages to all the nodes - # about the very same node. - # Thanks to gossip this information should propagate across all the - # cluster in a matter of seconds. - first = false - @nodes.each{|n| - if !first then first = n.info; next; end # Skip the first node - n.r.cluster("meet",first[:host],first[:port]) - } - end - - def yes_or_die(msg) - print "#{msg} (type 'yes' to accept): " - STDOUT.flush - if !(STDIN.gets.chomp.downcase == "yes") - puts "Aborting..." - exit 1 - end - end - - 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,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 - - # 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 -end - -COMMANDS={ - "create" => ["create_cluster_cmd", -2, "host1:port host2:port ... hostN:port"], - "check" => ["check_cluster_cmd", 2, "host:port"], - "reshard" => ["reshard_cluster_cmd", 2, "host:port"] -} - -# Sanity check -if ARGV.length == 0 - puts "Usage: redis-trib " - puts - COMMANDS.each{|k,v| - puts " #{k.ljust(20)} #{v[2]}" - } - puts - exit 1 -end - -rt = RedisTrib.new -cmd_spec = COMMANDS[ARGV[0].downcase] -if !cmd_spec - puts "Unknown redis-trib subcommand '#{ARGV[0]}'" - exit 1 -end -rt.check_arity(cmd_spec[1],ARGV.length) - -# Dispatch -rt.send(cmd_spec[0]) -- 2.47.2