From ef43e2b325ed7885d9332a33db3d285a492eb67f Mon Sep 17 00:00:00 2001 From: James Allen Date: Tue, 7 Jun 2016 18:38:32 +0100 Subject: [PATCH] Configure backend database as redis cluster --- services/document-updater/.gitignore | 5 + .../app/coffee/RedisBackend.coffee | 8 +- .../config/settings.defaults.coffee | 7 +- services/document-updater/package.json | 1 + .../redis_cluster/7000/redis.conf | 5 + .../redis_cluster/7001/redis.conf | 5 + .../redis_cluster/7002/redis.conf | 5 + .../redis_cluster/7003/redis.conf | 5 + .../redis_cluster/7004/redis.conf | 5 + .../redis_cluster/7005/redis.conf | 5 + .../redis_cluster/redis-cluster.sh | 23 + .../redis_cluster/redis-trib.rb | 1696 +++++++++++++++++ 12 files changed, 1765 insertions(+), 5 deletions(-) create mode 100644 services/document-updater/redis_cluster/7000/redis.conf create mode 100644 services/document-updater/redis_cluster/7001/redis.conf create mode 100644 services/document-updater/redis_cluster/7002/redis.conf create mode 100644 services/document-updater/redis_cluster/7003/redis.conf create mode 100644 services/document-updater/redis_cluster/7004/redis.conf create mode 100644 services/document-updater/redis_cluster/7005/redis.conf create mode 100755 services/document-updater/redis_cluster/redis-cluster.sh create mode 100755 services/document-updater/redis_cluster/redis-trib.rb diff --git a/services/document-updater/.gitignore b/services/document-updater/.gitignore index a477cfd66c..ad21f261b4 100644 --- a/services/document-updater/.gitignore +++ b/services/document-updater/.gitignore @@ -46,3 +46,8 @@ test/acceptance/js/* forever/ **.swp + +# Redis cluster +**/appendonly.aof +**/dump.rdb +**/nodes.conf diff --git a/services/document-updater/app/coffee/RedisBackend.coffee b/services/document-updater/app/coffee/RedisBackend.coffee index 3c795e827c..e242beaec2 100644 --- a/services/document-updater/app/coffee/RedisBackend.coffee +++ b/services/document-updater/app/coffee/RedisBackend.coffee @@ -1,5 +1,4 @@ Settings = require "settings-sharelatex" -redis = require("redis-sharelatex") async = require "async" class Client @@ -78,7 +77,12 @@ module.exports = client_configs.primary = true client_configs = [client_configs] clients = client_configs.map (config) -> - rclient: redis.createClient(config) + if config.cluster? + Redis = require("ioredis") + rclient = new Redis.Cluster(config.cluster) + else + rclient = require("redis-sharelatex").createClient(config) + rclient: rclient key_schema: config.key_schema primary: config.primary return new Client(clients) \ No newline at end of file diff --git a/services/document-updater/config/settings.defaults.coffee b/services/document-updater/config/settings.defaults.coffee index 38306d5d79..9a5c6d91f2 100755 --- a/services/document-updater/config/settings.defaults.coffee +++ b/services/document-updater/config/settings.defaults.coffee @@ -34,9 +34,10 @@ module.exports = pendingUpdates: ({doc_id}) -> "PendingUpdates:#{doc_id}" docsInProject: ({project_id}) -> "DocsIn:#{project_id}" }, { - port: "6380" - host: "localhost" - password:"" + cluster: [{ + port: "7000" + host: "localhost" + }] key_schema: blockingKey: ({doc_id}) -> "Blocking:{#{doc_id}}" docLines: ({doc_id}) -> "doclines:{#{doc_id}}" diff --git a/services/document-updater/package.json b/services/document-updater/package.json index d27b16271c..e91f3ff7c8 100644 --- a/services/document-updater/package.json +++ b/services/document-updater/package.json @@ -10,6 +10,7 @@ "async": "^2.0.0-rc.5", "coffee-script": "1.4.0", "express": "3.3.4", + "ioredis": "^2.0.1", "logger-sharelatex": "git+https://github.com/sharelatex/logger-sharelatex.git#v1.0.0", "lynx": "0.0.11", "metrics-sharelatex": "git+https://github.com/sharelatex/metrics-sharelatex.git#v1.0.0", diff --git a/services/document-updater/redis_cluster/7000/redis.conf b/services/document-updater/redis_cluster/7000/redis.conf new file mode 100644 index 0000000000..10dc9cb85c --- /dev/null +++ b/services/document-updater/redis_cluster/7000/redis.conf @@ -0,0 +1,5 @@ +port 7000 +cluster-enabled yes +cluster-config-file nodes.conf +cluster-node-timeout 5000 +appendonly yes \ No newline at end of file diff --git a/services/document-updater/redis_cluster/7001/redis.conf b/services/document-updater/redis_cluster/7001/redis.conf new file mode 100644 index 0000000000..10df3f5a5c --- /dev/null +++ b/services/document-updater/redis_cluster/7001/redis.conf @@ -0,0 +1,5 @@ +port 7001 +cluster-enabled yes +cluster-config-file nodes.conf +cluster-node-timeout 5000 +appendonly yes \ No newline at end of file diff --git a/services/document-updater/redis_cluster/7002/redis.conf b/services/document-updater/redis_cluster/7002/redis.conf new file mode 100644 index 0000000000..d16f4e2a7f --- /dev/null +++ b/services/document-updater/redis_cluster/7002/redis.conf @@ -0,0 +1,5 @@ +port 7002 +cluster-enabled yes +cluster-config-file nodes.conf +cluster-node-timeout 5000 +appendonly yes \ No newline at end of file diff --git a/services/document-updater/redis_cluster/7003/redis.conf b/services/document-updater/redis_cluster/7003/redis.conf new file mode 100644 index 0000000000..f54103d83f --- /dev/null +++ b/services/document-updater/redis_cluster/7003/redis.conf @@ -0,0 +1,5 @@ +port 7003 +cluster-enabled yes +cluster-config-file nodes.conf +cluster-node-timeout 5000 +appendonly yes \ No newline at end of file diff --git a/services/document-updater/redis_cluster/7004/redis.conf b/services/document-updater/redis_cluster/7004/redis.conf new file mode 100644 index 0000000000..8b3af0834b --- /dev/null +++ b/services/document-updater/redis_cluster/7004/redis.conf @@ -0,0 +1,5 @@ +port 7004 +cluster-enabled yes +cluster-config-file nodes.conf +cluster-node-timeout 5000 +appendonly yes \ No newline at end of file diff --git a/services/document-updater/redis_cluster/7005/redis.conf b/services/document-updater/redis_cluster/7005/redis.conf new file mode 100644 index 0000000000..ce0b227510 --- /dev/null +++ b/services/document-updater/redis_cluster/7005/redis.conf @@ -0,0 +1,5 @@ +port 7005 +cluster-enabled yes +cluster-config-file nodes.conf +cluster-node-timeout 5000 +appendonly yes \ No newline at end of file diff --git a/services/document-updater/redis_cluster/redis-cluster.sh b/services/document-updater/redis_cluster/redis-cluster.sh new file mode 100755 index 0000000000..e25359bcf7 --- /dev/null +++ b/services/document-updater/redis_cluster/redis-cluster.sh @@ -0,0 +1,23 @@ +#!/bin/sh + +(cd 7000 && redis-server redis.conf) & +PID1="$!" + +(cd 7001 && redis-server redis.conf) & +PID2="$!" + +(cd 7002 && redis-server redis.conf) & +PID3="$!" + +(cd 7003 && redis-server redis.conf) & +PID4="$!" + +(cd 7004 && redis-server redis.conf) & +PID5="$!" + +(cd 7005 && redis-server redis.conf) & +PID6="$!" + +trap "kill $PID1 $PID2 $PID3 $PID4 $PID5 $PID6" exit INT TERM + +wait \ No newline at end of file diff --git a/services/document-updater/redis_cluster/redis-trib.rb b/services/document-updater/redis_cluster/redis-trib.rb new file mode 100755 index 0000000000..ccb0551586 --- /dev/null +++ b/services/document-updater/redis_cluster/redis-trib.rb @@ -0,0 +1,1696 @@ +#!/usr/bin/env ruby + +# TODO (temporary here, we'll move this into the Github issues once +# redis-trib initial implementation is completed). +# +# - 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 condition 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 = 16384 +MigrateDefaultTimeout = 60000 +MigrateDefaultPipeline = 10 +RebalanceDefaultThreshold = 2 + +$verbose = false + +def xputs(s) + case s[0..2] + when ">>>" + color="29;1" + when "[ER" + color="31;1" + when "[WA" + color="31;1" + when "[OK" + color="32" + when "[FA","***" + color="33" + else + color=nil + end + + color = nil if ENV['TERM'] != "xterm" + print "\033[#{color}m" if color + print s + print "\033[0m" if color + print "\n" +end + +class ClusterNode + def initialize(addr) + s = addr.split("@")[0].split(":") + if s.length < 2 + puts "Invalid IP or Port (given as #{addr}) - use IP:Port format" + exit 1 + end + port = s.pop # removes port from split array + ip = s.join(":") # if s.length > 1 here, it's IPv6, so restore address + @r = nil + @info = {} + @info[:host] = ip + @info[:port] = port + @info[:slots] = {} + @info[:migrating] = {} + @info[:importing] = {} + @info[:replicate] = false + @dirty = false # True if we need to flush slots info into node. + @friends = [] + end + + def friends + @friends + end + + def slots + @info[:slots] + end + + def has_flag?(flag) + @info[:flags].index(flag) + end + + def to_s + "#{@info[:host]}:#{@info[:port]}" + end + + def connect(o={}) + return if @r + print "Connecting to node #{self}: " if $verbose + STDOUT.flush + begin + @r = Redis.new(:host => @info[:host], :port => @info[:port], :timeout => 60) + @r.ping + rescue + xputs "[ERR] Sorry, can't connect to node #{self}" + exit 1 if o[:abort] + @r = nil + end + xputs "OK" if $verbose + end + + def assert_cluster + info = @r.info + if !info["cluster_enabled"] || info["cluster_enabled"].to_i == 0 + xputs "[ERR] 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']) + xputs "[ERR] Node #{self} is not empty. Either the node already knows other nodes (check with CLUSTER NODES) 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,master_id,ping_sent,ping_recv,config_epoch,link_status = split[0..6] + slots = split[8..-1] + info = { + :name => name, + :addr => addr, + :flags => flags.split(","), + :replicate => master_id, + :ping_sent => ping_sent.to_i, + :ping_recv => ping_recv.to_i, + :link_status => link_status + } + info[:replicate] = false if master_id == "-" + + if info[:flags].index("myself") + @info = @info.merge(info) + @info[:slots] = {} + slots.each{|s| + if s[0..0] == '[' + if s.index("->-") # Migrating + slot,dst = s[1..-1].split("->-") + @info[:migrating][slot.to_i] = dst + elsif s.index("-<-") # Importing + slot,src = s[1..-1].split("-<-") + @info[:importing][slot.to_i] = src + end + 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 set_as_replica(node_id) + @info[:replicate] = node_id + @dirty = true + end + + def flush_node_config + return if !@dirty + if @info[:replicate] + begin + @r.cluster("replicate",@info[:replicate]) + rescue + # If the cluster did not already joined it is possible that + # the slave does not know the master node yet. So on errors + # we return ASAP leaving the dirty flag set, to flush the + # config later. + return + end + else + new = [] + @info[:slots].each{|s,val| + if val == :new + new << s + @info[:slots][s] = true + end + } + @r.cluster("addslots",*new) + end + @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 adjacent 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 adjacent 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(",") + + role = self.has_flag?("master") ? "M" : "S" + + if self.info[:replicate] and @dirty + is = "S: #{self.info[:name]} #{self.to_s}" + else + is = "#{role}: #{self.info[:name]} #{self.to_s}\n"+ + " slots:#{slots} (#{self.slots.length} slots) "+ + "#{(self.info[:flags]-["myself"]).join(",")}" + end + if self.info[:replicate] + is += "\n replicates #{info[:replicate]}" + elsif self.has_flag?("master") && self.info[:replicas] + is += "\n #{info[:replicas].length} additional replica(s)" + end + is + end + + # Return a single string representing nodes and associated slots. + # TODO: remove slaves from config when slaves will be handled + # by Redis Cluster. + def get_config_signature + config = [] + @r.cluster("nodes").each_line{|l| + s = l.split + slots = s[8..-1].select {|x| x[0..0] != "["} + next if slots.length == 0 + config << s[0]+":"+(slots.sort.join(",")) + } + config.sort.join("|") + end + + def info + @info + end + + def is_dirty? + @dirty + end + + def r + @r + end +end + +class RedisTrib + def initialize + @nodes = [] + @fix = false + @errors = [] + @timeout = MigrateDefaultTimeout + 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)) + xputs "[ERR] Wrong number of arguments for specified sub command" + exit 1 + end + end + + def add_node(node) + @nodes << node + end + + def reset_nodes + @nodes = [] + end + + def cluster_error(msg) + @errors << msg + xputs msg + end + + # Return the node with the specified ID or Nil. + def get_node_by_name(name) + @nodes.each{|n| + return n if n.info[:name] == name.downcase + } + return nil + end + + # Like get_node_by_name but the specified name can be just the first + # part of the node ID as long as the prefix in unique across the + # cluster. + def get_node_by_abbreviated_name(name) + l = name.length + candidates = [] + @nodes.each{|n| + if n.info[:name][0...l] == name.downcase + candidates << n + end + } + return nil if candidates.length != 1 + candidates[0] + end + + # This function returns the master that has the least number of replicas + # in the cluster. If there are multiple masters with the same smaller + # number of replicas, one at random is returned. + def get_master_with_least_replicas + masters = @nodes.select{|n| n.has_flag? "master"} + sorted = masters.sort{|a,b| + a.info[:replicas].length <=> b.info[:replicas].length + } + sorted[0] + end + + def check_cluster(opt={}) + xputs ">>> Performing Cluster Check (using node #{@nodes[0]})" + show_nodes if !opt[:quiet] + check_config_consistency + check_open_slots + check_slots_coverage + end + + def show_cluster_info + masters = 0 + keys = 0 + @nodes.each{|n| + if n.has_flag?("master") + puts "#{n} (#{n.info[:name][0...8]}...) -> #{n.r.dbsize} keys | #{n.slots.length} slots | "+ + "#{n.info[:replicas].length} slaves." + masters += 1 + keys += n.r.dbsize + end + } + xputs "[OK] #{keys} keys in #{masters} masters." + keys_per_slot = sprintf("%.2f",keys/16384.0) + puts "#{keys_per_slot} keys per slot on average." + end + + # Merge slots of every known node. If the resulting slots are equal + # to ClusterHashSlots, then all slots are served. + def covered_slots + slots = {} + @nodes.each{|n| + slots = slots.merge(n.slots) + } + slots + end + + def check_slots_coverage + xputs ">>> Check slots coverage..." + slots = covered_slots + if slots.length == ClusterHashSlots + xputs "[OK] All #{ClusterHashSlots} slots covered." + else + cluster_error \ + "[ERR] Not all #{ClusterHashSlots} slots are covered by nodes." + fix_slots_coverage if @fix + end + end + + def check_open_slots + xputs ">>> Check for open slots..." + open_slots = [] + @nodes.each{|n| + if n.info[:migrating].size > 0 + cluster_error \ + "[WARNING] Node #{n} has slots in migrating state (#{n.info[:migrating].keys.join(",")})." + open_slots += n.info[:migrating].keys + end + if n.info[:importing].size > 0 + cluster_error \ + "[WARNING] Node #{n} has slots in importing state (#{n.info[:importing].keys.join(",")})." + open_slots += n.info[:importing].keys + end + } + open_slots.uniq! + if open_slots.length > 0 + xputs "[WARNING] The following slots are open: #{open_slots.join(",")}" + end + if @fix + open_slots.each{|slot| fix_open_slot slot} + end + end + + def nodes_with_keys_in_slot(slot) + nodes = [] + @nodes.each{|n| + next if n.has_flag?("slave") + nodes << n if n.r.cluster("getkeysinslot",slot,1).length > 0 + } + nodes + end + + def fix_slots_coverage + not_covered = (0...ClusterHashSlots).to_a - covered_slots.keys + xputs ">>> Fixing slots coverage..." + xputs "List of not covered slots: " + not_covered.join(",") + + # For every slot, take action depending on the actual condition: + # 1) No node has keys for this slot. + # 2) A single node has keys for this slot. + # 3) Multiple nodes have keys for this slot. + slots = {} + not_covered.each{|slot| + nodes = nodes_with_keys_in_slot(slot) + slots[slot] = nodes + xputs "Slot #{slot} has keys in #{nodes.length} nodes: #{nodes.join(", ")}" + } + + none = slots.select {|k,v| v.length == 0} + single = slots.select {|k,v| v.length == 1} + multi = slots.select {|k,v| v.length > 1} + + # Handle case "1": keys in no node. + if none.length > 0 + xputs "The folowing uncovered slots have no keys across the cluster:" + xputs none.keys.join(",") + yes_or_die "Fix these slots by covering with a random node?" + none.each{|slot,nodes| + node = @nodes.sample + xputs ">>> Covering slot #{slot} with #{node}" + node.r.cluster("addslots",slot) + } + end + + # Handle case "2": keys only in one node. + if single.length > 0 + xputs "The folowing uncovered slots have keys in just one node:" + puts single.keys.join(",") + yes_or_die "Fix these slots by covering with those nodes?" + single.each{|slot,nodes| + xputs ">>> Covering slot #{slot} with #{nodes[0]}" + nodes[0].r.cluster("addslots",slot) + } + end + + # Handle case "3": keys in multiple nodes. + if multi.length > 0 + xputs "The folowing uncovered slots have keys in multiple nodes:" + xputs multi.keys.join(",") + yes_or_die "Fix these slots by moving keys into a single node?" + multi.each{|slot,nodes| + target = get_node_with_most_keys_in_slot(nodes,slot) + xputs ">>> Covering slot #{slot} moving keys to #{target}" + + target.r.cluster('addslots',slot) + target.r.cluster('setslot',slot,'stable') + nodes.each{|src| + next if src == target + # Set the source node in 'importing' state (even if we will + # actually migrate keys away) in order to avoid receiving + # redirections for MIGRATE. + src.r.cluster('setslot',slot,'importing',target.info[:name]) + move_slot(src,target,slot,:dots=>true,:fix=>true,:cold=>true) + src.r.cluster('setslot',slot,'stable') + } + } + end + end + + # Return the owner of the specified slot + def get_slot_owners(slot) + owners = [] + @nodes.each{|n| + next if n.has_flag?("slave") + n.slots.each{|s,_| + owners << n if s == slot + } + } + owners + end + + # Return the node, among 'nodes' with the greatest number of keys + # in the specified slot. + def get_node_with_most_keys_in_slot(nodes,slot) + best = nil + best_numkeys = 0 + @nodes.each{|n| + next if n.has_flag?("slave") + numkeys = n.r.cluster("countkeysinslot",slot) + if numkeys > best_numkeys || best == nil + best = n + best_numkeys = numkeys + end + } + return best + end + + # Slot 'slot' was found to be in importing or migrating state in one or + # more nodes. This function fixes this condition by migrating keys where + # it seems more sensible. + def fix_open_slot(slot) + puts ">>> Fixing open slot #{slot}" + + # Try to obtain the current slot owner, according to the current + # nodes configuration. + owners = get_slot_owners(slot) + owner = owners[0] if owners.length == 1 + + migrating = [] + importing = [] + @nodes.each{|n| + next if n.has_flag? "slave" + if n.info[:migrating][slot] + migrating << n + elsif n.info[:importing][slot] + importing << n + elsif n.r.cluster("countkeysinslot",slot) > 0 && n != owner + xputs "*** Found keys about slot #{slot} in node #{n}!" + importing << n + end + } + puts "Set as migrating in: #{migrating.join(",")}" + puts "Set as importing in: #{importing.join(",")}" + + # If there is no slot owner, set as owner the slot with the biggest + # number of keys, among the set of migrating / importing nodes. + if !owner + xputs ">>> Nobody claims ownership, selecting an owner..." + owner = get_node_with_most_keys_in_slot(@nodes,slot) + + # If we still don't have an owner, we can't fix it. + if !owner + xputs "[ERR] Can't select a slot owner. Impossible to fix." + exit 1 + end + + # Use ADDSLOTS to assign the slot. + puts "*** Configuring #{owner} as the slot owner" + owner.r.cluster("setslot",slot,"stable") + owner.r.cluster("addslots",slot) + # Make sure this information will propagate. Not strictly needed + # since there is no past owner, so all the other nodes will accept + # whatever epoch this node will claim the slot with. + owner.r.cluster("bumpepoch") + + # Remove the owner from the list of migrating/importing + # nodes. + migrating.delete(owner) + importing.delete(owner) + end + + # If there are multiple owners of the slot, we need to fix it + # so that a single node is the owner and all the other nodes + # are in importing state. Later the fix can be handled by one + # of the base cases above. + # + # Note that this case also covers multiple nodes having the slot + # in migrating state, since migrating is a valid state only for + # slot owners. + if owners.length > 1 + owner = get_node_with_most_keys_in_slot(owners,slot) + owners.each{|n| + next if n == owner + n.r.cluster('delslots',slot) + n.r.cluster('setslot',slot,'importing',owner.info[:name]) + importing.delete(n) # Avoid duplciates + importing << n + } + owner.r.cluster('bumpepoch') + end + + # Case 1: The slot is in migrating state in one slot, and in + # importing state in 1 slot. That's trivial to address. + if migrating.length == 1 && importing.length == 1 + move_slot(migrating[0],importing[0],slot,:dots=>true,:fix=>true) + # Case 2: There are multiple nodes that claim the slot as importing, + # they probably got keys about the slot after a restart so opened + # the slot. In this case we just move all the keys to the owner + # according to the configuration. + elsif migrating.length == 0 && importing.length > 0 + xputs ">>> Moving all the #{slot} slot keys to its owner #{owner}" + importing.each {|node| + next if node == owner + move_slot(node,owner,slot,:dots=>true,:fix=>true,:cold=>true) + xputs ">>> Setting #{slot} as STABLE in #{node}" + node.r.cluster("setslot",slot,"stable") + } + # Case 3: There are no slots claiming to be in importing state, but + # there is a migrating node that actually don't have any key. We + # can just close the slot, probably a reshard interrupted in the middle. + elsif importing.length == 0 && migrating.length == 1 && + migrating[0].r.cluster("getkeysinslot",slot,10).length == 0 + migrating[0].r.cluster("setslot",slot,"stable") + else + xputs "[ERR] Sorry, Redis-trib can't fix this slot yet (work in progress). Slot is set as migrating in #{migrating.join(",")}, as importing in #{importing.join(",")}, owner is #{owner}" + end + end + + # Check if all the nodes agree about the cluster configuration + def check_config_consistency + if !is_config_consistent? + cluster_error "[ERR] Nodes don't agree about configuration!" + else + xputs "[OK] All nodes agree about slots configuration." + end + end + + def is_config_consistent? + signatures=[] + @nodes.each{|n| + signatures << n.get_config_signature + } + return signatures.uniq.length == 1 + end + + def wait_cluster_join + print "Waiting for the cluster to join" + while !is_config_consistent? + print "." + STDOUT.flush + sleep 1 + end + print "\n" + end + + def alloc_slots + nodes_count = @nodes.length + masters_count = @nodes.length / (@replicas+1) + masters = [] + + # The first step is to split instances by IP. This is useful as + # we'll try to allocate master nodes in different physical machines + # (as much as possible) and to allocate slaves of a given master in + # different physical machines as well. + # + # This code assumes just that if the IP is different, than it is more + # likely that the instance is running in a different physical host + # or at least a different virtual machine. + ips = {} + @nodes.each{|n| + ips[n.info[:host]] = [] if !ips[n.info[:host]] + ips[n.info[:host]] << n + } + + # Select master instances + puts "Using #{masters_count} masters:" + interleaved = [] + stop = false + while not stop do + # Take one node from each IP until we run out of nodes + # across every IP. + ips.each do |ip,nodes| + if nodes.empty? + # if this IP has no remaining nodes, check for termination + if interleaved.length == nodes_count + # stop when 'interleaved' has accumulated all nodes + stop = true + next + end + else + # else, move one node from this IP to 'interleaved' + interleaved.push nodes.shift + end + end + end + + masters = interleaved.slice!(0, masters_count) + nodes_count -= masters.length + + masters.each{|m| puts m} + + # Alloc slots on masters + slots_per_node = ClusterHashSlots.to_f / masters_count + first = 0 + cursor = 0.0 + masters.each_with_index{|n,masternum| + last = (cursor+slots_per_node-1).round + if last > ClusterHashSlots || masternum == masters.length-1 + last = ClusterHashSlots-1 + end + last = first if last < first # Min step is 1. + n.add_slots first..last + first = last+1 + cursor += slots_per_node + } + + # Select N replicas for every master. + # We try to split the replicas among all the IPs with spare nodes + # trying to avoid the host where the master is running, if possible. + # + # Note we loop two times. The first loop assigns the requested + # number of replicas to each master. The second loop assigns any + # remaining instances as extra replicas to masters. Some masters + # may end up with more than their requested number of replicas, but + # all nodes will be used. + assignment_verbose = false + + [:requested,:unused].each do |assign| + masters.each do |m| + assigned_replicas = 0 + while assigned_replicas < @replicas + break if nodes_count == 0 + if assignment_verbose + if assign == :requested + puts "Requesting total of #{@replicas} replicas " \ + "(#{assigned_replicas} replicas assigned " \ + "so far with #{nodes_count} total remaining)." + elsif assign == :unused + puts "Assigning extra instance to replication " \ + "role too (#{nodes_count} remaining)." + end + end + + # Return the first node not matching our current master + node = interleaved.find{|n| n.info[:host] != m.info[:host]} + + # If we found a node, use it as a best-first match. + # Otherwise, we didn't find a node on a different IP, so we + # go ahead and use a same-IP replica. + if node + slave = node + interleaved.delete node + else + slave = interleaved.shift + end + slave.set_as_replica(m.info[:name]) + nodes_count -= 1 + assigned_replicas += 1 + puts "Adding replica #{slave} to #{m}" + + # If we are in the "assign extra nodes" loop, + # we want to assign one extra replica to each + # master before repeating masters. + # This break lets us assign extra replicas to masters + # in a round-robin way. + break if assign == :unused + end + end + end + end + + def flush_nodes_config + @nodes.each{|n| + n.flush_node_config + } + end + + def show_nodes + @nodes.each{|n| + xputs n.info_string + } + end + + # Redis Cluster config epoch collision resolution code is able to eventually + # set a different epoch to each node after a new cluster is created, but + # it is slow compared to assign a progressive config epoch to each node + # before joining the cluster. However we do just a best-effort try here + # since if we fail is not a problem. + def assign_config_epoch + config_epoch = 1 + @nodes.each{|n| + begin + n.r.cluster("set-config-epoch",config_epoch) + rescue + end + config_epoch += 1 + } + 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") + xputs "*** Aborting..." + exit 1 + end + end + + def load_cluster_info_from_node(nodeaddr) + node = ClusterNode.new(nodeaddr) + node.connect(:abort => true) + node.assert_cluster + node.load_info(:getfriends => true) + add_node(node) + node.friends.each{|f| + next if f[:flags].index("noaddr") || + f[:flags].index("disconnected") || + f[:flags].index("fail") + fnode = ClusterNode.new(f[:addr]) + fnode.connect() + next if !fnode.r + begin + fnode.load_info() + add_node(fnode) + rescue => e + xputs "[ERR] Unable to load info for node #{fnode}" + end + } + populate_nodes_replicas_info + end + + # This function is called by load_cluster_info_from_node in order to + # add additional information to every node as a list of replicas. + def populate_nodes_replicas_info + # Start adding the new field to every node. + @nodes.each{|n| + n.info[:replicas] = [] + } + + # Populate the replicas field using the replicate field of slave + # nodes. + @nodes.each{|n| + if n.info[:replicate] + master = get_node_by_name(n.info[:replicate]) + if !master + xputs "*** WARNING: #{n} claims to be slave of unknown node ID #{n.info[:replicate]}." + else + master.info[:replicas] << n + end + end + } + 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 instances 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 + + # Move slots between source and target nodes using MIGRATE. + # + # Options: + # :verbose -- Print a dot for every moved key. + # :fix -- We are moving in the context of a fix. Use REPLACE. + # :cold -- Move keys without opening slots / reconfiguring the nodes. + # :update -- Update nodes.info[:slots] for source/target nodes. + # :quiet -- Don't print info messages. + def move_slot(source,target,slot,o={}) + o = {:pipeline => MigrateDefaultPipeline}.merge(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. + if !o[:quiet] + print "Moving slot #{slot} from #{source} to #{target}: " + STDOUT.flush + end + + if !o[:cold] + target.r.cluster("setslot",slot,"importing",source.info[:name]) + source.r.cluster("setslot",slot,"migrating",target.info[:name]) + end + # Migrate all the keys from source to target using the MIGRATE command + while true + keys = source.r.cluster("getkeysinslot",slot,o[:pipeline]) + break if keys.length == 0 + begin + source.r.client.call(["migrate",target.info[:host],target.info[:port],"",0,@timeout,:keys,*keys]) + rescue => e + if o[:fix] && e.to_s =~ /BUSYKEY/ + xputs "*** Target key exists. Replacing it for FIX." + source.r.client.call(["migrate",target.info[:host],target.info[:port],"",0,@timeout,:replace,:keys,*keys]) + else + puts "" + xputs "[ERR] Calling MIGRATE: #{e}" + exit 1 + end + end + print "."*keys.length if o[:dots] + STDOUT.flush + end + + puts if !o[:quiet] + # Set the new node as the owner of the slot in all the known nodes. + if !o[:cold] + @nodes.each{|n| + next if n.has_flag?("slave") + n.r.cluster("setslot",slot,"node",target.info[:name]) + } + end + + # Update the node logical config + if o[:update] then + source.info[:slots].delete(slot) + target.info[:slots][slot] = true + end + end + + # redis-trib subcommands implementations. + + def check_cluster_cmd(argv,opt) + load_cluster_info_from_node(argv[0]) + check_cluster + end + + def info_cluster_cmd(argv,opt) + load_cluster_info_from_node(argv[0]) + show_cluster_info + end + + def rebalance_cluster_cmd(argv,opt) + opt = { + 'pipeline' => MigrateDefaultPipeline, + 'threshold' => RebalanceDefaultThreshold + }.merge(opt) + + # Load nodes info before parsing options, otherwise we can't + # handle --weight. + load_cluster_info_from_node(argv[0]) + + # Options parsing + threshold = opt['threshold'].to_i + autoweights = opt['auto-weights'] + weights = {} + opt['weight'].each{|w| + fields = w.split("=") + node = get_node_by_abbreviated_name(fields[0]) + if !node || !node.has_flag?("master") + puts "*** No such master node #{fields[0]}" + exit 1 + end + weights[node.info[:name]] = fields[1].to_f + } if opt['weight'] + useempty = opt['use-empty-masters'] + + # Assign a weight to each node, and compute the total cluster weight. + total_weight = 0 + nodes_involved = 0 + @nodes.each{|n| + if n.has_flag?("master") + next if !useempty && n.slots.length == 0 + n.info[:w] = weights[n.info[:name]] ? weights[n.info[:name]] : 1 + total_weight += n.info[:w] + nodes_involved += 1 + end + } + + # Check cluster, only proceed if it looks sane. + check_cluster(:quiet => true) + if @errors.length != 0 + puts "*** Please fix your cluster problems before rebalancing" + exit 1 + end + + # Calculate the slots balance for each node. It's the number of + # slots the node should lose (if positive) or gain (if negative) + # in order to be balanced. + threshold = opt['threshold'].to_f + threshold_reached = false + @nodes.each{|n| + if n.has_flag?("master") + next if !n.info[:w] + expected = ((ClusterHashSlots.to_f / total_weight) * + n.info[:w]).to_i + n.info[:balance] = n.slots.length - expected + # Compute the percentage of difference between the + # expected number of slots and the real one, to see + # if it's over the threshold specified by the user. + over_threshold = false + if threshold > 0 + if n.slots.length > 0 + err_perc = (100-(100.0*expected/n.slots.length)).abs + over_threshold = true if err_perc > threshold + elsif expected > 0 + over_threshold = true + end + end + threshold_reached = true if over_threshold + end + } + if !threshold_reached + xputs "*** No rebalancing needed! All nodes are within the #{threshold}% threshold." + return + end + + # Only consider nodes we want to change + sn = @nodes.select{|n| + n.has_flag?("master") && n.info[:w] + } + + # Because of rounding, it is possible that the balance of all nodes + # summed does not give 0. Make sure that nodes that have to provide + # slots are always matched by nodes receiving slots. + total_balance = sn.map{|x| x.info[:balance]}.reduce{|a,b| a+b} + while total_balance > 0 + sn.each{|n| + if n.info[:balance] < 0 && total_balance > 0 + n.info[:balance] -= 1 + total_balance -= 1 + end + } + end + + # Sort nodes by their slots balance. + sn = sn.sort{|a,b| + a.info[:balance] <=> b.info[:balance] + } + + xputs ">>> Rebalancing across #{nodes_involved} nodes. Total weight = #{total_weight}" + + if $verbose + sn.each{|n| + puts "#{n} balance is #{n.info[:balance]} slots" + } + end + + # Now we have at the start of the 'sn' array nodes that should get + # slots, at the end nodes that must give slots. + # We take two indexes, one at the start, and one at the end, + # incrementing or decrementing the indexes accordingly til we + # find nodes that need to get/provide slots. + dst_idx = 0 + src_idx = sn.length - 1 + + while dst_idx < src_idx + dst = sn[dst_idx] + src = sn[src_idx] + numslots = [dst.info[:balance],src.info[:balance]].map{|n| + n.abs + }.min + + if numslots > 0 + puts "Moving #{numslots} slots from #{src} to #{dst}" + + # Actaully move the slots. + reshard_table = compute_reshard_table([src],numslots) + if reshard_table.length != numslots + xputs "*** Assertio failed: Reshard table != number of slots" + exit 1 + end + if opt['simulate'] + print "#"*reshard_table.length + else + reshard_table.each{|e| + move_slot(e[:source],dst,e[:slot], + :quiet=>true, + :dots=>false, + :update=>true, + :pipeline=>opt['pipeline']) + print "#" + STDOUT.flush + } + end + puts + end + + # Update nodes balance. + dst.info[:balance] += numslots + src.info[:balance] -= numslots + dst_idx += 1 if dst.info[:balance] == 0 + src_idx -= 1 if src.info[:balance] == 0 + end + end + + def fix_cluster_cmd(argv,opt) + @fix = true + @timeout = opt['timeout'].to_i if opt['timeout'] + + load_cluster_info_from_node(argv[0]) + check_cluster + end + + def reshard_cluster_cmd(argv,opt) + opt = {'pipeline' => MigrateDefaultPipeline}.merge(opt) + + load_cluster_info_from_node(argv[0]) + check_cluster + if @errors.length != 0 + puts "*** Please fix your cluster problems before resharding" + exit 1 + end + + @timeout = opt['timeout'].to_i if opt['timeout'].to_i + + # Get number of slots + if opt['slots'] + numslots = opt['slots'].to_i + else + numslots = 0 + while numslots <= 0 or numslots > ClusterHashSlots + print "How many slots do you want to move (from 1 to #{ClusterHashSlots})? " + numslots = STDIN.gets.to_i + end + end + + # Get the target instance + if opt['to'] + target = get_node_by_name(opt['to']) + if !target || target.has_flag?("slave") + xputs "*** The specified node is not known or not a master, please retry." + exit 1 + end + else + target = nil + while not target + print "What is the receiving node ID? " + target = get_node_by_name(STDIN.gets.chop) + if !target || target.has_flag?("slave") + xputs "*** The specified node is not known or not a master, please retry." + target = nil + end + end + end + + # Get the source instances + sources = [] + if opt['from'] + opt['from'].split(',').each{|node_id| + if node_id == "all" + sources = "all" + break + end + src = get_node_by_name(node_id) + if !src || src.has_flag?("slave") + xputs "*** The specified node is not known or is not a master, please retry." + exit 1 + end + sources << src + } + else + xputs "Please enter all the source node IDs." + xputs " Type 'all' to use all the nodes as source nodes for the hash slots." + xputs " 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" + break + elsif line == "all" + sources = "all" + break + elsif !src || src.has_flag?("slave") + xputs "*** The specified node is not known or is not a master, please retry." + elsif src.info[:name] == target.info[:name] + xputs "*** It is not possible to use the target node as source node." + else + sources << src + end + end + end + + if sources.length == 0 + puts "*** No source nodes given, operation aborted" + exit 1 + end + + # Handle soures == all. + if sources == "all" + sources = [] + @nodes.each{|n| + next if n.info[:name] == target.info[:name] + next if n.has_flag?("slave") + sources << n + } + end + + # Check if the destination node is the same of any source nodes. + if sources.index(target) + xputs "*** Target node is also listed among the source nodes!" + exit 1 + 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) + if !opt['yes'] + print "Do you want to proceed with the proposed reshard plan (yes/no)? " + yesno = STDIN.gets.chop + exit(1) if (yesno != "yes") + end + reshard_table.each{|e| + move_slot(e[:source],target,e[:slot], + :dots=>true, + :pipeline=>opt['pipeline']) + } + end + + # This is an helper function for create_cluster_cmd that verifies if + # the number of nodes and the specified replicas have a valid configuration + # where there are at least three master nodes and enough replicas per node. + def check_create_parameters + masters = @nodes.length/(@replicas+1) + if masters < 3 + puts "*** ERROR: Invalid configuration for cluster creation." + puts "*** Redis Cluster requires at least 3 master nodes." + puts "*** This is not possible with #{@nodes.length} nodes and #{@replicas} replicas per node." + puts "*** At least #{3*(@replicas+1)} nodes are required." + exit 1 + end + end + + def create_cluster_cmd(argv,opt) + opt = {'replicas' => 0}.merge(opt) + @replicas = opt['replicas'].to_i + + xputs ">>> Creating cluster" + argv[0..-1].each{|n| + node = ClusterNode.new(n) + node.connect(:abort => true) + node.assert_cluster + node.load_info + node.assert_empty + add_node(node) + } + check_create_parameters + xputs ">>> Performing hash slots allocation on #{@nodes.length} nodes..." + alloc_slots + show_nodes + yes_or_die "Can I set the above configuration?" + flush_nodes_config + xputs ">>> Nodes configuration updated" + xputs ">>> Assign a different config epoch to each node" + assign_config_epoch + xputs ">>> Sending CLUSTER MEET messages to join the cluster" + join_cluster + # Give one second for the join to start, in order to avoid that + # wait_cluster_join will find all the nodes agree about the config as + # they are still empty with unassigned slots. + sleep 1 + wait_cluster_join + flush_nodes_config # Useful for the replicas + check_cluster + end + + def addnode_cluster_cmd(argv,opt) + xputs ">>> Adding node #{argv[0]} to cluster #{argv[1]}" + + # Check the existing cluster + load_cluster_info_from_node(argv[1]) + check_cluster + + # If --master-id was specified, try to resolve it now so that we + # abort before starting with the node configuration. + if opt['slave'] + if opt['master-id'] + master = get_node_by_name(opt['master-id']) + if !master + xputs "[ERR] No such master ID #{opt['master-id']}" + end + else + master = get_master_with_least_replicas + xputs "Automatically selected master #{master}" + end + end + + # Add the new node + new = ClusterNode.new(argv[0]) + new.connect(:abort => true) + new.assert_cluster + new.load_info + new.assert_empty + first = @nodes.first.info + add_node(new) + + # Send CLUSTER MEET command to the new node + xputs ">>> Send CLUSTER MEET to node #{new} to make it join the cluster." + new.r.cluster("meet",first[:host],first[:port]) + + # Additional configuration is needed if the node is added as + # a slave. + if opt['slave'] + wait_cluster_join + xputs ">>> Configure node as replica of #{master}." + new.r.cluster("replicate",master.info[:name]) + end + xputs "[OK] New node added correctly." + end + + def delnode_cluster_cmd(argv,opt) + id = argv[1].downcase + xputs ">>> Removing node #{id} from cluster #{argv[0]}" + + # Load cluster information + load_cluster_info_from_node(argv[0]) + + # Check if the node exists and is not empty + node = get_node_by_name(id) + + if !node + xputs "[ERR] No such node ID #{id}" + exit 1 + end + + if node.slots.length != 0 + xputs "[ERR] Node #{node} is not empty! Reshard data away and try again." + exit 1 + end + + # Send CLUSTER FORGET to all the nodes but the node to remove + xputs ">>> Sending CLUSTER FORGET messages to the cluster..." + @nodes.each{|n| + next if n == node + if n.info[:replicate] && n.info[:replicate].downcase == id + # Reconfigure the slave to replicate with some other node + master = get_master_with_least_replicas + xputs ">>> #{n} as replica of #{master}" + n.r.cluster("replicate",master.info[:name]) + end + n.r.cluster("forget",argv[1]) + } + + # Finally shutdown the node + xputs ">>> SHUTDOWN the node." + node.r.shutdown + end + + def set_timeout_cluster_cmd(argv,opt) + timeout = argv[1].to_i + if timeout < 100 + puts "Setting a node timeout of less than 100 milliseconds is a bad idea." + exit 1 + end + + # Load cluster information + load_cluster_info_from_node(argv[0]) + ok_count = 0 + err_count = 0 + + # Send CLUSTER FORGET to all the nodes but the node to remove + xputs ">>> Reconfiguring node timeout in every cluster node..." + @nodes.each{|n| + begin + n.r.config("set","cluster-node-timeout",timeout) + n.r.config("rewrite") + ok_count += 1 + xputs "*** New timeout set for #{n}" + rescue => e + puts "ERR setting node-timeot for #{n}: #{e}" + err_count += 1 + end + } + xputs ">>> New node timeout set. #{ok_count} OK, #{err_count} ERR." + end + + def call_cluster_cmd(argv,opt) + cmd = argv[1..-1] + cmd[0] = cmd[0].upcase + + # Load cluster information + load_cluster_info_from_node(argv[0]) + xputs ">>> Calling #{cmd.join(" ")}" + @nodes.each{|n| + begin + res = n.r.send(*cmd) + puts "#{n}: #{res}" + rescue => e + puts "#{n}: #{e}" + end + } + end + + def import_cluster_cmd(argv,opt) + source_addr = opt['from'] + xputs ">>> Importing data from #{source_addr} to cluster #{argv[1]}" + use_copy = opt['copy'] + use_replace = opt['replace'] + + # Check the existing cluster. + load_cluster_info_from_node(argv[0]) + check_cluster + + # Connect to the source node. + xputs ">>> Connecting to the source Redis instance" + src_host,src_port = source_addr.split(":") + source = Redis.new(:host =>src_host, :port =>src_port) + if source.info['cluster_enabled'].to_i == 1 + xputs "[ERR] The source node should not be a cluster node." + end + xputs "*** Importing #{source.dbsize} keys from DB 0" + + # Build a slot -> node map + slots = {} + @nodes.each{|n| + n.slots.each{|s,_| + slots[s] = n + } + } + + # Use SCAN to iterate over the keys, migrating to the + # right node as needed. + cursor = nil + while cursor != 0 + cursor,keys = source.scan(cursor, :count => 1000) + cursor = cursor.to_i + keys.each{|k| + # Migrate keys using the MIGRATE command. + slot = key_to_slot(k) + target = slots[slot] + print "Migrating #{k} to #{target}: " + STDOUT.flush + begin + cmd = ["migrate",target.info[:host],target.info[:port],k,0,@timeout] + cmd << :copy if use_copy + cmd << :replace if use_replace + source.client.call(cmd) + rescue => e + puts e + else + puts "OK" + end + } + end + end + + def help_cluster_cmd(argv,opt) + show_help + exit 0 + end + + # Parse the options for the specific command "cmd". + # Returns an hash populate with option => value pairs, and the index of + # the first non-option argument in ARGV. + def parse_options(cmd) + idx = 1 ; # Current index into ARGV + options={} + while idx < ARGV.length && ARGV[idx][0..1] == '--' + if ARGV[idx][0..1] == "--" + option = ARGV[idx][2..-1] + idx += 1 + + # --verbose is a global option + if option == "verbose" + $verbose = true + next + end + + if ALLOWED_OPTIONS[cmd] == nil || ALLOWED_OPTIONS[cmd][option] == nil + puts "Unknown option '#{option}' for command '#{cmd}'" + exit 1 + end + if ALLOWED_OPTIONS[cmd][option] != false + value = ARGV[idx] + idx += 1 + else + value = true + end + + # If the option is set to [], it's a multiple arguments + # option. We just queue every new value into an array. + if ALLOWED_OPTIONS[cmd][option] == [] + options[option] = [] if !options[option] + options[option] << value + else + options[option] = value + end + else + # Remaining arguments are not options. + break + end + end + + # Enforce mandatory options + if ALLOWED_OPTIONS[cmd] + ALLOWED_OPTIONS[cmd].each {|option,val| + if !options[option] && val == :required + puts "Option '--#{option}' is required "+ \ + "for subcommand '#{cmd}'" + exit 1 + end + } + end + return options,idx + end +end + +################################################################################# +# Libraries +# +# We try to don't depend on external libs since this is a critical part +# of Redis Cluster. +################################################################################# + +# This is the CRC16 algorithm used by Redis Cluster to hash keys. +# Implementation according to CCITT standards. +# +# This is actually the XMODEM CRC 16 algorithm, using the +# following parameters: +# +# Name : "XMODEM", also known as "ZMODEM", "CRC-16/ACORN" +# Width : 16 bit +# Poly : 1021 (That is actually x^16 + x^12 + x^5 + 1) +# Initialization : 0000 +# Reflect Input byte : False +# Reflect Output CRC : False +# Xor constant to output CRC : 0000 +# Output for "123456789" : 31C3 + +module RedisClusterCRC16 + def RedisClusterCRC16.crc16(bytes) + crc = 0 + bytes.each_byte{|b| + crc = ((crc<<8) & 0xffff) ^ XMODEMCRC16Lookup[((crc>>8)^b) & 0xff] + } + crc + end + +private + XMODEMCRC16Lookup = [ + 0x0000,0x1021,0x2042,0x3063,0x4084,0x50a5,0x60c6,0x70e7, + 0x8108,0x9129,0xa14a,0xb16b,0xc18c,0xd1ad,0xe1ce,0xf1ef, + 0x1231,0x0210,0x3273,0x2252,0x52b5,0x4294,0x72f7,0x62d6, + 0x9339,0x8318,0xb37b,0xa35a,0xd3bd,0xc39c,0xf3ff,0xe3de, + 0x2462,0x3443,0x0420,0x1401,0x64e6,0x74c7,0x44a4,0x5485, + 0xa56a,0xb54b,0x8528,0x9509,0xe5ee,0xf5cf,0xc5ac,0xd58d, + 0x3653,0x2672,0x1611,0x0630,0x76d7,0x66f6,0x5695,0x46b4, + 0xb75b,0xa77a,0x9719,0x8738,0xf7df,0xe7fe,0xd79d,0xc7bc, + 0x48c4,0x58e5,0x6886,0x78a7,0x0840,0x1861,0x2802,0x3823, + 0xc9cc,0xd9ed,0xe98e,0xf9af,0x8948,0x9969,0xa90a,0xb92b, + 0x5af5,0x4ad4,0x7ab7,0x6a96,0x1a71,0x0a50,0x3a33,0x2a12, + 0xdbfd,0xcbdc,0xfbbf,0xeb9e,0x9b79,0x8b58,0xbb3b,0xab1a, + 0x6ca6,0x7c87,0x4ce4,0x5cc5,0x2c22,0x3c03,0x0c60,0x1c41, + 0xedae,0xfd8f,0xcdec,0xddcd,0xad2a,0xbd0b,0x8d68,0x9d49, + 0x7e97,0x6eb6,0x5ed5,0x4ef4,0x3e13,0x2e32,0x1e51,0x0e70, + 0xff9f,0xefbe,0xdfdd,0xcffc,0xbf1b,0xaf3a,0x9f59,0x8f78, + 0x9188,0x81a9,0xb1ca,0xa1eb,0xd10c,0xc12d,0xf14e,0xe16f, + 0x1080,0x00a1,0x30c2,0x20e3,0x5004,0x4025,0x7046,0x6067, + 0x83b9,0x9398,0xa3fb,0xb3da,0xc33d,0xd31c,0xe37f,0xf35e, + 0x02b1,0x1290,0x22f3,0x32d2,0x4235,0x5214,0x6277,0x7256, + 0xb5ea,0xa5cb,0x95a8,0x8589,0xf56e,0xe54f,0xd52c,0xc50d, + 0x34e2,0x24c3,0x14a0,0x0481,0x7466,0x6447,0x5424,0x4405, + 0xa7db,0xb7fa,0x8799,0x97b8,0xe75f,0xf77e,0xc71d,0xd73c, + 0x26d3,0x36f2,0x0691,0x16b0,0x6657,0x7676,0x4615,0x5634, + 0xd94c,0xc96d,0xf90e,0xe92f,0x99c8,0x89e9,0xb98a,0xa9ab, + 0x5844,0x4865,0x7806,0x6827,0x18c0,0x08e1,0x3882,0x28a3, + 0xcb7d,0xdb5c,0xeb3f,0xfb1e,0x8bf9,0x9bd8,0xabbb,0xbb9a, + 0x4a75,0x5a54,0x6a37,0x7a16,0x0af1,0x1ad0,0x2ab3,0x3a92, + 0xfd2e,0xed0f,0xdd6c,0xcd4d,0xbdaa,0xad8b,0x9de8,0x8dc9, + 0x7c26,0x6c07,0x5c64,0x4c45,0x3ca2,0x2c83,0x1ce0,0x0cc1, + 0xef1f,0xff3e,0xcf5d,0xdf7c,0xaf9b,0xbfba,0x8fd9,0x9ff8, + 0x6e17,0x7e36,0x4e55,0x5e74,0x2e93,0x3eb2,0x0ed1,0x1ef0 + ] +end + +# Turn a key name into the corrisponding Redis Cluster slot. +def key_to_slot(key) + # Only hash what is inside {...} if there is such a pattern in the key. + # Note that the specification requires the content that is between + # the first { and the first } after the first {. If we found {} without + # nothing in the middle, the whole key is hashed as usually. + s = key.index "{" + if s + e = key.index "}",s+1 + if e && e != s+1 + key = key[s+1..e-1] + end + end + RedisClusterCRC16.crc16(key) % 16384 +end + +################################################################################# +# Definition of commands +################################################################################# + +COMMANDS={ + "create" => ["create_cluster_cmd", -2, "host1:port1 ... hostN:portN"], + "check" => ["check_cluster_cmd", 2, "host:port"], + "info" => ["info_cluster_cmd", 2, "host:port"], + "fix" => ["fix_cluster_cmd", 2, "host:port"], + "reshard" => ["reshard_cluster_cmd", 2, "host:port"], + "rebalance" => ["rebalance_cluster_cmd", -2, "host:port"], + "add-node" => ["addnode_cluster_cmd", 3, "new_host:new_port existing_host:existing_port"], + "del-node" => ["delnode_cluster_cmd", 3, "host:port node_id"], + "set-timeout" => ["set_timeout_cluster_cmd", 3, "host:port milliseconds"], + "call" => ["call_cluster_cmd", -3, "host:port command arg arg .. arg"], + "import" => ["import_cluster_cmd", 2, "host:port"], + "help" => ["help_cluster_cmd", 1, "(show this help)"] +} + +ALLOWED_OPTIONS={ + "create" => {"replicas" => true}, + "add-node" => {"slave" => false, "master-id" => true}, + "import" => {"from" => :required, "copy" => false, "replace" => false}, + "reshard" => {"from" => true, "to" => true, "slots" => true, "yes" => false, "timeout" => true, "pipeline" => true}, + "rebalance" => {"weight" => [], "auto-weights" => false, "use-empty-masters" => false, "timeout" => true, "simulate" => false, "pipeline" => true, "threshold" => true}, + "fix" => {"timeout" => MigrateDefaultTimeout}, +} + +def show_help + puts "Usage: redis-trib \n\n" + COMMANDS.each{|k,v| + o = "" + puts " #{k.ljust(15)} #{v[2]}" + if ALLOWED_OPTIONS[k] + ALLOWED_OPTIONS[k].each{|optname,has_arg| + puts " --#{optname}" + (has_arg ? " " : "") + } + end + } + puts "\nFor check, fix, reshard, del-node, set-timeout you can specify the host and port of any working node in the cluster.\n" +end + +# Sanity check +if ARGV.length == 0 + show_help + 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 + +# Parse options +cmd_options,first_non_option = rt.parse_options(ARGV[0].downcase) +rt.check_arity(cmd_spec[1],ARGV.length-(first_non_option-1)) + +# Dispatch +rt.send(cmd_spec[0],ARGV[first_non_option..-1],cmd_options) \ No newline at end of file