Read-only flag removed from PUBLISH command.

[redis.git] / src / redis-trib.rb

diff --git a/src/redis-trib.rb b/src/redis-trib.rb
index 083f9740ffa2a9bd2fc09eddc59e66b2e1e14fbc..473e492292912d60cbafbcaf917d24c61183b36a 100755 (executable)
--- a/src/redis-trib.rb
+++ b/src/redis-trib.rb
@@ -1,5 +1,26 @@
 #!/usr/bin/env ruby
 
 #!/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'
 
 require 'rubygems'
 require 'redis'
 


@@ -74,7 +95,9 @@ class ClusterNode
         nodes = @r.cluster("nodes").split("\n")
         nodes.each{|n|
             # name addr flags role ping_sent ping_recv link_status slots
         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,
             info = {
                 :name => name,
                 :addr => addr,


@@ -87,8 +110,10 @@ class ClusterNode
             if info[:flags].index("myself")
                 @info = @info.merge(info)
                 @info[:slots] = {}
             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
                         start,stop = s.split("-")
                         self.add_slots((start.to_i)..(stop.to_i))
                     else


@@ -164,7 +189,7 @@ class ClusterNode
             x.count == 1 ? x.first.to_s : "#{x.first}-#{x.last}"
         }.join(",")
 
             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
     end
 
     def info


@@ -287,10 +312,23 @@ class RedisTrib
     # instance.
     def compute_reshard_table(sources,numslots)
         moved = []
     # 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.
             # 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}
             s.slots.keys.sort[(0...n)].each{|slot|
                 if moved.length < numslots
                     moved << {:source => s, :slot => slot}


@@ -302,7 +340,32 @@ class RedisTrib
 
     def show_reshard_table(table)
         table.each{|e|
 
     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
 


@@ -322,7 +385,7 @@ class RedisTrib
         end
         numslots = 0
         while numslots <= 0 or numslots > 4096
         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
             numslots = STDIN.gets.to_i
         end
         target = nil


@@ -368,7 +431,14 @@ class RedisTrib
         puts "  Destination node:"
         puts "    #{target.info_string}"
         reshard_table = compute_reshard_table(sources,numslots)
         puts "  Destination node:"
         puts "    #{target.info_string}"
         reshard_table = compute_reshard_table(sources,numslots)

+        puts "  Resharding plan:"

         show_reshard_table(reshard_table)
         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
     end
 
     def create_cluster_cmd


@@ -377,6 +447,7 @@ class RedisTrib
             node = ClusterNode.new(n)
             node.connect(:abort => true)
             node.assert_cluster
             node = ClusterNode.new(n)
             node.connect(:abort => true)
             node.assert_cluster

+            node.load_info

             node.assert_empty
             add_node(node)
         }
             node.assert_empty
             add_node(node)
         }