Merge branch 'sentinel' of https://github.com/antirez/redis-rb into antirez-sentinel

djanowski · djanowski · commit e07eefc502a4 · 2014-10-29T10:43:51.000-03:00
diff --git a/README.md b/README.md
@@ -85,6 +85,27 @@ available on [rdoc.info][rdoc].
 
 [rdoc]: http://rdoc.info/github/redis/redis-rb/
 
+## Sentinel support
+
+Redis-rb is able to optionally fetch the current master address using
+[Redis Sentinel](http://redis.io/topics/sentinel). The new
+[Sentinel handshake protocol](http://redis.io/topics/sentinel-clients)
+is supported, so redis-rb when used with Sentinel will automatically connect
+to the new master after a failover, assuming you use a Recent version of
+Redis 2.8.
+
+To connect using Sentinel, use:
+
+```ruby
+Sentinels = [{:host => "127.0.0.1", :port => 26380},
+             {:host => "127.0.0.1", :port => 26381}]
+r = Redis.new(:url => "sentinel://mymaster", :sentinels => Sentinels, :role => :master)
+```
+
+* The master name, that identifies a group of Redis instances composed of a master and one or more slaves, is specified in the url parameter (`mymaster` in the example).
+* It is possible to optionally provide a role. The allowed roles are `master` and `slave`, and the default is `master`. When the role is `slave` redis-rb will try to fetch a list of slaves and will connect to a random slave.
+* When using the Sentinel support you need to specify a list of Sentinels to connect to. The list does not need to enumerate all your Sentinel instances, but a few so that if one is down redis-rb will try the next one. The client is able to remember the last Sentinel that was able to reply correctly and will use it for the next requests.
+
 ## Storing objects
 
 Redis only stores strings as values. If you want to store an object, you
diff --git a/examples/consistency.rb b/examples/consistency.rb
@@ -0,0 +1,114 @@
+# This file implements a simple consistency test for Redis-rb (or any other
+# Redis environment if you pass a different client object) where a client
+# writes to the database using INCR in order to increment keys, but actively
+# remember the value the key should have. Before every write a read is performed
+# to check if the value in the database matches the value expected.
+#
+# In this way this program can check for lost writes, or acknowledged writes
+# that were executed.
+#
+# Copyright (C) 2013-2014 Salvatore Sanfilippo <antirez@gmail.com>
+#
+# Permission is hereby granted, free of charge, to any person obtaining
+# a copy of this software and associated documentation files (the
+# "Software"), to deal in the Software without restriction, including
+# without limitation the rights to use, copy, modify, merge, publish,
+# distribute, sublicense, and/or sell copies of the Software, and to
+# permit persons to whom the Software is furnished to do so, subject to
+# the following conditions:
+# 
+# The above copyright notice and this permission notice shall be
+# included in all copies or substantial portions of the Software.
+# 
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+require 'redis'
+
+class ConsistencyTester
+    def initialize(redis)
+        @r = redis
+        @working_set = 10000
+        @keyspace = 100000
+        @writes = 0
+        @reads = 0
+        @failed_writes = 0
+        @failed_reads = 0
+        @lost_writes = 0
+        @not_ack_writes = 0
+        @delay = 0
+        @cached = {} # We take our view of data stored in the DB.
+        @prefix = [Process.pid.to_s,Time.now.usec,@r.object_id,""].join("|")
+        @errtime = {}
+    end
+
+    def genkey
+        # Write more often to a small subset of keys
+        ks = rand() > 0.5 ? @keyspace : @working_set
+        @prefix+"key_"+rand(ks).to_s
+    end
+
+    def check_consistency(key,value)
+        expected = @cached[key]
+        return if !expected  # We lack info about previous state.
+        if expected > value
+            @lost_writes += expected-value
+        elsif expected < value
+            @not_ack_writes += value-expected
+        end
+    end
+
+    def puterr(msg)
+        if !@errtime[msg] || Time.now.to_i != @errtime[msg]
+            puts msg
+        end
+        @errtime[msg] = Time.now.to_i
+    end
+
+    def test
+        last_report = Time.now.to_i
+        while true
+            # Read
+            key = genkey
+            begin
+                val = @r.get(key)
+                check_consistency(key,val.to_i)
+                @reads += 1
+            rescue => e
+                puterr "Reading: #{e.to_s}"
+                @failed_reads += 1
+            end
+
+            # Write
+            begin
+                @cached[key] = @r.incr(key).to_i
+                @writes += 1
+            rescue => e
+                puterr "Writing: #{e.to_s}"
+                @failed_writes += 1
+            end
+
+            # Report
+            sleep @delay
+            if Time.now.to_i != last_report
+                report = "#{@reads} R (#{@failed_reads} err) | " +
+                         "#{@writes} W (#{@failed_writes} err) | "
+                report += "#{@lost_writes} lost | " if @lost_writes > 0
+                report += "#{@not_ack_writes} noack | " if @not_ack_writes > 0
+                last_report = Time.now.to_i
+                puts report
+            end
+        end
+    end
+end
+
+Sentinels = [{:host => "127.0.0.1", :port => 26380},
+             {:host => "127.0.0.1", :port => 26381}]
+r = Redis.new(:url => "sentinel://mymaster", :sentinels => Sentinels, :role => :master)
+tester = ConsistencyTester.new(r)
+tester.test
diff --git a/examples/sentinel.rb b/examples/sentinel.rb
@@ -0,0 +1,19 @@
+require 'redis'
+
+Sentinels = [{:host => "127.0.0.1", :port => 26379},
+             {:host => "127.0.0.1", :port => 26380}]
+r = Redis.new(:url => "sentinel://mymaster", :sentinels => Sentinels, :role => :master)
+
+# Set keys into a loop.
+#
+# The example traps errors so that you can actually try to failover while
+# running the script to see redis-rb reconfiguring.
+(0..1000000).each{|i|
+    begin
+        r.set(i,i)
+        puts i
+    rescue
+        puts "ERROR #{i}"
+    end
+    sleep(0.01)
+}
diff --git a/lib/redis/client.rb b/lib/redis/client.rb
@@ -85,6 +85,25 @@ def connect
         establish_connection
         call [:auth, password] if password
         call [:select, db] if db != 0
+        if @options[:sentinels]
+          # Check the instance is really of the role we are looking for.
+          # We can't assume the command is supported since it was
+          # introduced recently and this client should work with old stuff.
+          role = nil
+          begin
+            role = call [:role]
+          rescue
+            # Assume the test is passed if we can't get a reply from ROLE...
+            role = [@options[:role].to_s]
+          end
+
+          # Raise an error on role mismatch. TODO: we could do better, wait
+          # some time and connect again...
+          if role[0] != @options[:role].to_s
+            disconnect
+            raise ConnectionError, "Instance role mismatch, try again."
+          end
+        end
       end
 
       self
@@ -300,9 +319,60 @@ def logging(commands)
       end
     end
 
+    def set_addr_via_sentinel
+      responder = nil # The Sentinel that was able to reply
+
+      # Try one Sentinel after the other, using the list provided
+      # by the user.
+      @options[:sentinels].each{|sentinel|
+        begin
+          if !sentinel[:link]
+            sentinel[:link] = Redis.new(:host => sentinel[:host],
+                                        :port => sentinel[:port],
+                                        :timeout => 0.300)
+          end
+          if @options[:role] == :master
+            reply = sentinel[:link].client.call(["sentinel","get-master-addr-by-name",@options[:mastername]])
+            next if !reply
+            # Got it, set :host and :port
+            @options[:host] = reply[0]
+            @options[:port] = reply[1]
+            responder = sentinel
+            break
+          elsif @options[:role] == :slave
+            reply = sentinel[:link].client.call(["sentinel","slaves",@options[:mastername]])
+            slaves = []
+            reply.each{|slave|
+                slaves << Hash[*slave]
+            }
+            random_slave = slaves[rand(slaves.length)]
+            @options[:host] = random_slave['ip']
+            @options[:port] = random_slave['port']
+            responder = sentinel
+            break
+          else
+            raise ArgumentError, "Unknown instance role #{@options[:role]}"
+          end
+        rescue
+            next; # Try the next one on error
+        end
+      }
+
+      if responder
+        # If we were able to obtain the address, make sure to put the Sentinel
+        # that was able to reply as the first in the list.
+        @options[:sentinels].delete(responder)
+        @options[:sentinels].unshift(responder)
+      else
+        raise ConnectionError, "Unable to fetch #{@options[:role]} via Sentinel."
+      end
+    end
+
     def establish_connection
+      if @options[:sentinels]
+        set_addr_via_sentinel
+      end
       @connection = @options[:driver].connect(@options.dup)
-
     rescue TimeoutError
       raise CannotConnectError, "Timed out connecting to Redis on #{location}"
     rescue Errno::ECONNREFUSED
@@ -365,7 +435,7 @@ def _parse_options(options)
 
         if uri.scheme == "unix"
           defaults[:path]   = uri.path
-        else
+        elsif uri.scheme == "redis"
           # Require the URL to have at least a host
           raise ArgumentError, "invalid url" unless uri.host
 
@@ -374,6 +444,14 @@ def _parse_options(options)
           defaults[:port]     = uri.port if uri.port
           defaults[:password] = CGI.unescape(uri.password) if uri.password
           defaults[:db]       = uri.path[1..-1].to_i if uri.path
+        elsif uri.scheme == "sentinel"
+          defaults[:scheme]   = uri.scheme
+          defaults[:mastername] = uri.host
+          defaults[:role] = :master
+          defaults[:password] = CGI.unescape(uri.password) if uri.password
+          defaults[:db]       = uri.path[1..-1].to_i if uri.path
+        else
+          raise ArgumentError, "invalid uri scheme '#{uri.scheme}'"
         end
       end
 
@@ -383,10 +461,19 @@ def _parse_options(options)
       end
 
       if options[:path]
+        # Unix socket
         options[:scheme] = "unix"
         options.delete(:host)
         options.delete(:port)
+      elsif options[:mastername]
+        # Sentinel
+        options.delete(:host)
+        options.delete(:port)
+        if options[:sentinels].nil?
+          raise ArgumentError, "list of Sentinels required"
+        end
       else
+        # TCP socket
         options[:host] = options[:host].to_s
         options[:port] = options[:port].to_i
       end
